diff --git a/CHANGELOG.md b/CHANGELOG.md
index f578346..fb3797c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,11 +16,22 @@
   one-level family graph, and an SSE event stream with `after=` resume.
   Hidden tabs pause; SSE failure falls back to the existing 5s poll.
   Workspace detail shows the family tree and live event panes. Copy
-  buttons emit dry-run CLI only. Browser stays read-only; P2 channel
-  POST and P3 artifacts are not implemented. Family badges do not claim
+  buttons emit dry-run CLI only. Family badges do not claim
   cleanliness or origin binding. Event `facts` stay IDs, enums, counts,
   short hashes, or reason codes. HMAC `after=` binds a row digest so a
   replaced same-seq row is `EVENT_CURSOR_INVALID`.
+- Console P2: overlay family channel at
+  `.dyro/families/<parent>/channel.jsonl` plus `acks.json`, dual-written
+  with a `signal` event under the same lock. CLI `line post` / `inbox` /
+  `ack` stay under `dyro line`. `next --format json` adds read-only
+  `family_unacked` and never emits `line post` or `line ack --yes`.
+  Console channel pane shows full family history; the browser POST
+  accepts only `decision` / `contract` / `ack` as `operator`. Artifact
+  bytes stay closed (P3). `/dyro-line-family` still does not post.
+  CLI `--to operator` uses the same default family as a broadcast
+  (`parent` or sender), not `F(sender)`. Channel pairing and ack treat
+  `msg_N` as per-family; HTTP ack is bound to the URL family, and CLI
+  `line ack` fail-closes when the same id exists in more than one family.
 - User slash Skill `dyro-line-family` (`/dyro-line-family`) preflights
   `line spawn` / `line merge` / `line sync` and prints one `--yes` command
   for the human. It does not execute the mutation, invent `--push`, or
diff --git a/docs/designs/console-v2-live-family-signals.md b/docs/designs/console-v2-live-family-signals.md
index 017610a..832fd5f 100644
--- a/docs/designs/console-v2-live-family-signals.md
+++ b/docs/designs/console-v2-live-family-signals.md
@@ -395,7 +395,7 @@ meta `surfaces` 增加 `events`（P1）与 `families`（P2）。页面按能力
 
 ### P1 · 事件 + 图
 
-P1 已在本 PR 落地（事件尾、`parent` 投影、一层家族图、SSE `after=`）。P2 / P3 仍未实现。
+P1 已在本 PR 落地（事件尾、`parent` 投影、一层家族图、SSE `after=`）。P3 仍未实现。
 
 - 写入并读取 `.dyro/events.jsonl`。
 - line DTO 投影 `parent`。
@@ -406,6 +406,8 @@ P1 已在本 PR 落地（事件尾、`parent` 投影、一层家族图、SSE `af
 
 ### P2 · 家族频道 + 人类模块
 
+P2 已落地：`channel.jsonl` / `acks`、`line post|inbox|ack`、`next family_unacked`、人类 POST `decision|contract|ack`。
+
 - 计算 `F(P)`，写入 `channel.jsonl` 与对应 `signal` 事件。
 - CLI：`line post` / `inbox` / `ack`。
 - `next` 与控制面读取未 ack。
diff --git a/src/dyro/cli.py b/src/dyro/cli.py
index 78abd8f..3fbca0d 100644
--- a/src/dyro/cli.py
+++ b/src/dyro/cli.py
@@ -102,6 +102,7 @@
 )
 from .evidence import build_execution_bundle, unpack_execution_bundle
 from .errors import DyroError, ValidationError
+from .families import ack_channel_message, family_unacked, list_inbox, post_channel_message
 from .home import (
     HomeTool,
     _record_for_root,
@@ -390,6 +391,36 @@ def _require_objective_yes(args: argparse.Namespace, label: str) -> None:
         )
 
 
+def _require_overlay_yes(args: argparse.Namespace, label: str) -> None:
+    if not args.yes and not args.dry_run:
+        raise DyroError(
+            f"{label} 会写入 overlay 家族信号；请先使用 --dry-run 检查，再加 --yes 执行"
+        )
+
+
+def _family_unacked_fields(config: Config) -> dict[str, object]:
+    try:
+        payload = family_unacked(config)
+    except (DyroError, OSError, UnicodeError):
+        payload = {"count": 0, "kind": "", "family": "", "summary": ""}
+    if not isinstance(payload, dict):
+        payload = {"count": 0, "kind": "", "family": "", "summary": ""}
+    return {"family_unacked": payload}
+
+
+def _print_family_unacked_attention(config: Config) -> None:
+    payload = _family_unacked_fields(config)["family_unacked"]
+    if isinstance(payload, dict) and payload.get("count"):
+        print("家族频道有未读信号")
+
+
+def _inbox_viewer(config: Config) -> str:
+    try:
+        return resolve_line(config, interactive=False).id
+    except (DyroError, ValidationError):
+        return ""
+
+
 def _objective_contract_from_args(args: argparse.Namespace, config: Config) -> str:
     if args.file:
         path = Path(args.file).expanduser()
@@ -2588,6 +2619,7 @@ def cmd_next(args: argparse.Namespace) -> None:
                 diagnostic_commands=[_briefing_command(args, config, "doctor")],
                 mutation_available=bootstrap_applicable,
                 findings=findings,
+                **_family_unacked_fields(config),
                 **_next_push_fields(config),
             )
             return
@@ -2600,6 +2632,7 @@ def cmd_next(args: argparse.Namespace) -> None:
                 "缺失仓库均已配置 remote，可运行："
                 + _briefing_command(args, config, "bootstrap", "--yes")
             )
+        _print_family_unacked_attention(config)
         _print_push_disclosure(config)
         return
     lines = list_lines(config, read_budget=budget)
@@ -2612,10 +2645,12 @@ def cmd_next(args: argparse.Namespace) -> None:
                 summary="Profile 已就绪，但还没有开发线。",
                 commands=[command],
                 mutation_available=True,
+                **_family_unacked_fields(config),
                 **_next_push_fields(config),
             )
             return
         print(f"Profile 已就绪，但还没有开发线。下一步：{command}")
+        _print_family_unacked_attention(config)
         _print_push_disclosure(config)
         return
     if not config.adapters and not installed_launchable_presets():
@@ -2627,6 +2662,7 @@ def cmd_next(args: argparse.Namespace) -> None:
                 commands=[],
                 mutation_available=False,
                 required_inputs=["agent_id", "agent_command"],
+                **_family_unacked_fields(config),
                 **_next_push_fields(config),
             )
             return
@@ -2635,6 +2671,7 @@ def cmd_next(args: argparse.Namespace) -> None:
             "安装本机 Agent 后运行 dyro start，或 "
             + _scoped_command(args, config, "agent", "add", "<id>", "--command", "…")
         )
+        _print_family_unacked_attention(config)
         _print_push_disclosure(config)
         return
     briefing, diagnostic_commands = _workspace_ready_briefing(
@@ -2655,6 +2692,7 @@ def cmd_next(args: argparse.Namespace) -> None:
                 _doctor_finding_payload(item, include_paths=False)
                 for item in missing_origin_failures
             ]
+        payload.update(_family_unacked_fields(config))
         payload.update(_next_push_fields(config))
         _print_control_plane_json("next_step", **payload)
         return
@@ -2662,9 +2700,11 @@ def cmd_next(args: argparse.Namespace) -> None:
         _print_doctor_finding(finding)
     if briefing is None:
         print("工作区已就绪。可用 dyro start 打开本机已安装的编码工具。")
+        _print_family_unacked_attention(config)
         _print_push_disclosure(config)
         return
     print(render_briefing_text(briefing))
+    _print_family_unacked_attention(config)
     _print_push_disclosure(config)
 
 
@@ -2791,6 +2831,70 @@ def cmd_line_sync(args: argparse.Namespace) -> None:
     )
 
 
+def cmd_line_post(args: argparse.Namespace) -> None:
+    config = _config(args)
+    _require_overlay_yes(args, "写入家族频道")
+    result = post_channel_message(
+        config,
+        sender=args.sender,
+        kind=args.kind,
+        body=args.body or "",
+        recipient=args.to or "",
+        family=args.family or "",
+        dry_run=args.dry_run,
+    )
+    if args.format == "json":
+        payload = dict(result)
+        payload["channel_kind"] = payload.pop("kind", "")
+        _print_control_plane_json("line_post", **payload)
+        return
+    target = result["to"] or "家族广播"
+    prefix = "DRY RUN: " if args.dry_run else ""
+    print(
+        f"{prefix}已从 {result['from']} 向 {target} 发送 {result['kind']} "
+        f"{result['id']}（家族 {result['family']}）"
+    )
+
+
+def cmd_line_inbox(args: argparse.Namespace) -> None:
+    config = _config(args)
+    inbox = list_inbox(
+        config,
+        family=args.family or "",
+        viewer=_inbox_viewer(config),
+        unacked=args.unacked,
+    )
+    if args.format == "json":
+        _print_control_plane_json("line_inbox", **inbox)
+        return
+    messages = inbox["messages"]
+    print(
+        f"家族 {inbox['family']} · 查看者 {inbox['viewer']} · 未读 {inbox['unacked']}"
+    )
+    if not messages:
+        print("没有可展示的家族信号")
+        return
+    for item in messages:
+        target = item["to"] or "广播"
+        flag = "未读" if not item["acked"] else "已读"
+        print(
+            f"{item['id']:10} {item['kind']:10} {item['from']} → {target}  {flag}  {item['body']}"
+        )
+
+
+def cmd_line_ack(args: argparse.Namespace) -> None:
+    config = _config(args)
+    _require_overlay_yes(args, "确认已读家族信号")
+    result = ack_channel_message(
+        config, args.id, family=args.family or "", dry_run=args.dry_run
+    )
+    if args.format == "json":
+        _print_control_plane_json("line_ack", **result)
+        return
+    prefix = "DRY RUN: " if args.dry_run else ""
+    print(f"{prefix}已确认已读 {result['id']}（家族 {result['family']}）")
+
+
 def cmd_hotfix_create(args: argparse.Namespace) -> None:
     _create_line(args, "hotfix")
 
@@ -4824,6 +4928,72 @@ def build_parser() -> argparse.ArgumentParser:
     )
     line_sync.add_argument("--yes", action="store_true")
     line_sync.set_defaults(func=cmd_line_sync)
+    line_post = line_sub.add_parser("post", help="向一层家族频道追加 overlay 信号")
+    line_post.add_argument("sender", metavar="LINE", help="发送者线 id；人类用 operator")
+    line_post.add_argument(
+        "--kind",
+        required=True,
+        choices=(
+            "contract",
+            "blocked",
+            "shipped",
+            "ask_sync",
+            "decision",
+            "artifact",
+            "retract",
+        ),
+        help="频道 kind；operator 只能发 decision 或 contract",
+    )
+    line_post.add_argument("--to", default="", help="定向接收者；省略则为家族广播")
+    line_post.add_argument("--body", default="", help="正文；retract 时填被撤回的 msg id")
+    line_post.add_argument(
+        "--family",
+        default="",
+        help="家族父线；operator 广播时必填",
+    )
+    line_post.add_argument(
+        "--dry-run",
+        dest="dry_run",
+        action="store_true",
+        default=argparse.SUPPRESS,
+        help="只打印计划，不写频道或事件（兼容全局 --dry-run）",
+    )
+    line_post.add_argument("--yes", action="store_true")
+    line_post.add_argument("--format", choices=("text", "json"), default="text")
+    line_post.set_defaults(func=cmd_line_post)
+    line_inbox = line_sub.add_parser("inbox", help="读取一层家族频道可见行")
+    line_inbox.add_argument("--family", default="", help="家族父线；省略则看当前线")
+    line_inbox.add_argument(
+        "--unacked",
+        action="store_true",
+        help="只列出操作者尚未 ack 的行",
+    )
+    line_inbox.add_argument(
+        "--dry-run",
+        dest="dry_run",
+        action="store_true",
+        default=argparse.SUPPRESS,
+        help="只读；不写频道（兼容全局 --dry-run）",
+    )
+    line_inbox.add_argument("--format", choices=("text", "json"), default="text")
+    line_inbox.set_defaults(func=cmd_line_inbox)
+    line_ack = line_sub.add_parser("ack", help="将一条家族信号标为人类已读")
+    line_ack.add_argument("id", help="频道行 id，例如 msg_1")
+    line_ack.add_argument(
+        "--family",
+        default="",
+        help="家族父线；同号出现在多个家族时必填",
+    )
+    line_ack.add_argument(
+        "--dry-run",
+        dest="dry_run",
+        action="store_true",
+        default=argparse.SUPPRESS,
+        help="只打印计划，不写 ack（兼容全局 --dry-run）",
+    )
+    line_ack.add_argument("--yes", action="store_true")
+    line_ack.add_argument("--format", choices=("text", "json"), default="text")
+    line_ack.set_defaults(func=cmd_line_ack)
 
     hotfix = sub.add_parser("hotfix", help="生产 Hotfix 开发线")
     hotfix_sub = hotfix.add_subparsers(dest="hotfix_command", required=True)
diff --git a/src/dyro/console/_inspect_worker.py b/src/dyro/console/_inspect_worker.py
index 01d9e73..e423643 100644
--- a/src/dyro/console/_inspect_worker.py
+++ b/src/dyro/console/_inspect_worker.py
@@ -241,6 +241,7 @@ def _decode_request(value: str) -> dict[str, object]:
         "alias",
         "after",
         "parent",
+        "filter",
         "target_root",
     }:
         raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
@@ -339,6 +340,25 @@ def main(argv: list[str] | None = None) -> int:
             if not isinstance(parent, str):
                 raise ConsoleOverviewError("FAMILY_PARENT_INVALID")
             payload = service.family(alias, parent)
+        elif operation == "channel":
+            alias = request.get("alias")
+            parent = request.get("parent")
+            if not isinstance(alias, str):
+                raise ConsoleOverviewError("WORKSPACE_ALIAS_INVALID")
+            if not isinstance(parent, str):
+                raise ConsoleOverviewError("FAMILY_PARENT_INVALID")
+            after = request.get("after")
+            if after is not None and not isinstance(after, str):
+                raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID")
+            filter_text = request.get("filter")
+            if filter_text is not None and not isinstance(filter_text, str):
+                raise ConsoleOverviewError("CHANNEL_FILTER_INVALID")
+            limit = request.get("limit", 50)
+            if type(limit) is not int:
+                raise ConsoleOverviewError("CHANNEL_LIMIT_INVALID")
+            payload = service.channel(
+                alias, parent, after=after, filter=filter_text, limit=limit
+            )
         else:
             raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
         return _response(payload)
diff --git a/src/dyro/console/assets.py b/src/dyro/console/assets.py
index 34338e1..386ccd5 100644
--- a/src/dyro/console/assets.py
+++ b/src/dyro/console/assets.py
@@ -27,13 +27,13 @@ class ConsoleAsset:
     ),
     "app.js": (
         "text/javascript; charset=utf-8",
-        "5ee284b3cd590ba114a64b1994ac79de64b7376bd38b5163f74176b704ac0113",
-        47380,
+        "194ca77cdfac96539dd391f43d6d14a5e15bf594ec21a280c886a21536124c23",
+        60441,
     ),
     "styles.css": (
         "text/css; charset=utf-8",
-        "dbf956c627803fe47ad910ec79a6a0763e6a40c993fa2e96e5ec79739fcd3432",
-        14676,
+        "810b196f7e72418d7253c1f2aa490f82fbd51a98a12d2db2d51e8303b4632f07",
+        15492,
     ),
 }
 
diff --git a/src/dyro/console/assets/app.js b/src/dyro/console/assets/app.js
index fc7ede7..e45a475 100644
--- a/src/dyro/console/assets/app.js
+++ b/src/dyro/console/assets/app.js
@@ -17,6 +17,13 @@ const state = {
   liveEdges: new Set(),
   sseFailed: false,
   familyParent: "",
+  channelItems: [],
+  channelCursor: "",
+  channelMembers: [],
+  channelView: "list",
+  channelKind: "",
+  channelFrom: "",
+  channelUnacked: false,
 };
 const HEALTH_LABELS = { healthy: "健康", degraded: "需关注", unavailable: "不可用" };
 const FRESHNESS_LABELS = { fresh: "读取完整", partial: "部分可读", stale: "待刷新" };
@@ -124,6 +131,10 @@ const ERROR_LABELS = {
   EVENT_CURSOR_INVALID: "事件游标已失效，已从头读取",
   EVENT_STREAM_UNAVAILABLE: "实时事件流不可用，已回退轮询",
   FAMILY_NOT_FOUND: "没有可展示的一层家族",
+  FAMILY_POST_FORBIDDEN: "人类模块不能发送这种家族信号",
+  FAMILY_POST_INVALID: "家族频道请求无效",
+  CHANNEL_CURSOR_INVALID: "频道游标已失效，已从头读取",
+  CHANNEL_BODY_INVALID: "家族信号正文不接受",
 };
 const EVENT_KIND_LABELS = {
   spawn: "子线已创建",
@@ -137,6 +148,16 @@ const EVENT_KIND_LABELS = {
   host_seed: "已写入 overlay",
   EVENT_REDACTED: "已脱敏事件",
 };
+const CHANNEL_KIND_LABELS = {
+  contract: "约定",
+  blocked: "阻塞",
+  shipped: "声称已交付",
+  ask_sync: "请求同步",
+  decision: "决定",
+  artifact: "产物",
+  retract: "撤回",
+};
+
 const UPDATE_KIND_LABELS = {
   none: "无已缓存更新",
   patch: "有补丁更新",
@@ -298,6 +319,26 @@ async function request(path, key) {
   return body;
 }
 
+async function requestWrite(path, payload) {
+  const response = await fetch(path, {
+    method: "POST",
+    headers: {
+      Authorization: `Bearer ${state.bearer}`,
+      "Content-Type": "application/json",
+    },
+    body: JSON.stringify(payload),
+    cache: "no-store",
+    credentials: "omit",
+  });
+  const body = await response.json().catch(() => null);
+  if (response.status === 401) throw new Error("SESSION_EXPIRED");
+  if (!response.ok || !body) {
+    const code = text(body && body.error && body.error.code) || "LOCAL_READ_UNAVAILABLE";
+    throw new Error(code);
+  }
+  return body;
+}
+
 function expireSession() {
   state.bearer = "";
   state.etags.clear();
@@ -841,6 +882,11 @@ function renderFamilyTree(alias, lines, tasks) {
         state.familyParent = parent;
         const tree = $("family-tree");
         if (tree) tree.replaceWith(renderFamilyGraph(alias, lines, parent, tasks));
+        resetChannelState();
+        loadChannel(alias, parent).catch((error) => {
+          if (error && error.message === "SESSION_EXPIRED") expireSession();
+        });
+        refreshFamilyUnread(alias, parent).catch(() => {});
       });
       nav.append(button);
     }
@@ -850,25 +896,23 @@ function renderFamilyTree(alias, lines, tasks) {
   return section;
 }
 
-function familyBadges(id, tasks) {
+function familyBadges(id, tasks, unread = 0) {
   const marks = element("p");
   marks.className = "family-badges";
+  const unreadBadge = element("span", `未读 ${count(unread)}`);
+  unreadBadge.className = "family-badge family-unread";
   if (id === "operator") {
-    marks.append(element("span", "未读 0"));
+    marks.append(unreadBadge);
     return marks;
   }
   const busy = lineInProgress(tasks, id);
-  // Git cleanliness and origin binding are not inspected in P1.
-  for (const label of [
-    "未检查",
-    "未检查",
-    busy ? "进行中" : "空闲",
-    "未读 0",
-  ]) {
+  // Git cleanliness and origin binding are not inspected. Unread is overlay-only.
+  for (const label of ["未检查", "未检查", busy ? "进行中" : "空闲"]) {
     const badge = element("span", label);
     badge.className = "family-badge";
     marks.append(badge);
   }
+  marks.append(unreadBadge);
   return marks;
 }
 
@@ -887,6 +931,7 @@ function renderFamilyGraph(alias, lines, parent, tasks) {
   for (const id of members) {
     const item = element("li");
     item.className = "family-node";
+    item.dataset.member = id;
     item.dataset.role = id === parent ? "parent" : id === "operator" ? "operator" : "child";
     const title = element("strong", id);
     const role = element(
@@ -1107,16 +1152,334 @@ function renderEventPane() {
   return section;
 }
 
-function renderChannelPane() {
+function resetChannelState() {
+  state.channelItems = [];
+  state.channelCursor = "";
+  state.channelMembers = [];
+}
+
+function channelFilterText() {
+  const parts = [];
+  if (state.channelUnacked) parts.push("unacked");
+  if (state.channelKind) parts.push(`kind:${state.channelKind}`);
+  if (state.channelFrom) parts.push(`from:${state.channelFrom}`);
+  return parts.join(",");
+}
+
+function describeChannelMessage(message) {
+  const kind = text(message && message.kind);
+  const label = kind === "artifact"
+    ? "产物尚未开放"
+    : displayLabel(kind, CHANNEL_KIND_LABELS);
+  const sender = text(message && message.from) || "未提供";
+  const recipient = text(message && message.to);
+  const who = recipient ? `${sender} → ${recipient}` : `${sender} → 广播`;
+  const when = text(message && message.at);
+  const local = when ? new Date(when).toLocaleString("zh-CN") : "";
+  const body = kind === "artifact" ? text(message && message.id) : text(message && message.body);
+  const flags = [
+    message && message.retracted ? "已撤回" : "",
+    message && message.acked ? "已读" : "未读",
+  ].filter(Boolean).join(" · ");
+  return [local, label, who, flags, body].filter(Boolean).join(" · ");
+}
+
+function appendChannelMessages(messages) {
+  if (!Array.isArray(messages) || !messages.length) return;
+  const seen = new Set(state.channelItems.map((item) => text(item.id)));
+  for (const message of messages) {
+    const id = text(message && message.id);
+    if (!id || seen.has(id)) continue;
+    state.channelItems.push(message);
+    seen.add(id);
+  }
+}
+
+function visibleChannelMessages() {
+  const items = [...state.channelItems];
+  if (state.channelView === "list") {
+    items.sort((left, right) => Number(Boolean(left.acked)) - Number(Boolean(right.acked)) || left.seq - right.seq);
+  } else {
+    items.sort((left, right) => left.seq - right.seq);
+  }
+  return items;
+}
+
+function renderChannelMessages() {
+  const list = $("channel-list");
+  if (!list) return;
+  list.replaceChildren();
+  const items = visibleChannelMessages();
+  if (!items.length) {
+    list.append(element("li", "还没有家族信号。"));
+    return;
+  }
+  const alias = state.detailAlias;
+  const parent = state.familyParent;
+  for (const message of items) {
+    const item = element("li");
+    item.className = "channel-item";
+    if (message.retracted) item.classList.add("retracted");
+    if (!message.acked) item.classList.add("unacked");
+    item.append(element("p", describeChannelMessage(message)));
+    if (text(message.kind) === "artifact") {
+      item.append(element("p", `产物尚未开放 · ${text(message.id)}`));
+    }
+    if (!message.acked && SAFE_ID.test(alias) && SAFE_ID.test(parent)) {
+      const ack = element("button", "标为已读");
+      ack.type = "button";
+      ack.className = "secondary";
+      ack.addEventListener("click", () => {
+        postHumanChannel(alias, parent, { kind: "ack", ack_id: text(message.id) }).catch((error) => {
+          if (error && error.message === "SESSION_EXPIRED") expireSession();
+        });
+      });
+      item.append(ack);
+    }
+    const sender = text(message.from);
+    const retractFrom = sender && sender !== "operator" ? sender : parent;
+    if (
+      alias &&
+      retractFrom &&
+      SAFE_ID.test(alias) &&
+      SAFE_ID.test(retractFrom) &&
+      text(message.kind) !== "retract" &&
+      text(message.id)
+    ) {
+      item.append(commandRow(
+        `dyro --workspace ${alias} --dry-run line post ${retractFrom} --kind retract --body ${text(message.id)}`,
+      ));
+    }
+    list.append(item);
+  }
+}
+
+function fillSelect(node, entries, selected) {
+  if (!node) return;
+  node.replaceChildren();
+  for (const [value, label] of entries) {
+    const option = document.createElement("option");
+    option.value = value;
+    option.textContent = label;
+    if (value === selected) option.selected = true;
+    node.append(option);
+  }
+}
+
+function renderChannelFilters(members) {
+  const bar = element("div");
+  bar.className = "channel-toolbar";
+  const views = element("div");
+  views.className = "channel-views";
+  views.setAttribute("role", "tablist");
+  for (const [id, label] of [["list", "列表"], ["timeline", "时间线"]]) {
+    const button = element("button", label);
+    button.type = "button";
+    button.className = "secondary";
+    if (state.channelView === id) button.setAttribute("aria-current", "true");
+    button.addEventListener("click", () => {
+      state.channelView = id;
+      for (const item of views.querySelectorAll("button")) {
+        if (item.textContent === label) item.setAttribute("aria-current", "true");
+        else item.removeAttribute("aria-current");
+      }
+      renderChannelMessages();
+    });
+    views.append(button);
+  }
+  const kind = element("select");
+  kind.id = "channel-filter-kind";
+  fillSelect(
+    kind,
+    [["", "全部 kind"], ...Object.entries(CHANNEL_KIND_LABELS)],
+    state.channelKind,
+  );
+  kind.addEventListener("change", () => {
+    state.channelKind = text(kind.value);
+    reloadOpenChannel();
+  });
+  const from = element("select");
+  from.id = "channel-filter-from";
+  fillSelect(
+    from,
+    [["", "全部发送者"], ...members.filter((id) => id).map((id) => [id, id])],
+    state.channelFrom,
+  );
+  from.addEventListener("change", () => {
+    state.channelFrom = text(from.value);
+    reloadOpenChannel();
+  });
+  const unacked = element("label");
+  const box = document.createElement("input");
+  box.type = "checkbox";
+  box.checked = state.channelUnacked;
+  box.addEventListener("change", () => {
+    state.channelUnacked = Boolean(box.checked);
+    reloadOpenChannel();
+  });
+  unacked.append(box, document.createTextNode("未读"));
+  bar.append(views, kind, from, unacked);
+  return bar;
+}
+
+function renderChannelCompose(alias, parent, members) {
+  const form = element("div");
+  form.className = "channel-compose";
+  form.append(element("p", "以 operator 身份发送。页面不能假扮开发线。"));
+  const kind = element("select");
+  kind.id = "channel-post-kind";
+  fillSelect(kind, [["decision", "决定"], ["contract", "约定"]], "decision");
+  const to = element("select");
+  to.id = "channel-post-to";
+  fillSelect(
+    to,
+    [["", "全家族"], ...members.filter((id) => id && id !== "operator").map((id) => [id, id])],
+    "",
+  );
+  const body = element("textarea");
+  body.id = "channel-post-body";
+  body.rows = 3;
+  body.maxLength = 2048;
+  const send = element("button", "发送决定");
+  send.type = "button";
+  kind.addEventListener("change", () => {
+    send.textContent = kind.value === "contract" ? "发送约定" : "发送决定";
+  });
+  send.addEventListener("click", () => {
+    postHumanChannel(alias, parent, {
+      kind: text(kind.value),
+      to: text(to.value),
+      body: text(body.value),
+    }).then(() => {
+      body.value = "";
+    }).catch((error) => {
+      if (error && error.message === "SESSION_EXPIRED") expireSession();
+    });
+  });
+  form.append(kind, to, body, send);
+  form.append(element("p", "页面不能 merge、push 或改任务状态。撤回只能复制 dry-run CLI。"));
+  return form;
+}
+
+function renderChannelPane(alias) {
   const section = element("section");
   section.className = "live-pane channel-pane";
   section.id = "channel-pane";
   section.append(element("h3", "频道"));
-  section.append(element("p", "尚未开放"));
+  if (!hasSurface("families")) {
+    section.append(element("p", "尚未开放"));
+    section.append(element("p", "产物尚未开放"));
+    return section;
+  }
+  section.append(element("p", "人类身份固定为 operator。"));
+  const parent = state.familyParent;
+  const members = state.channelMembers.length ? state.channelMembers : [];
+  section.append(renderChannelFilters(members));
+  const list = element("ul");
+  list.id = "channel-list";
+  list.className = "channel-list";
+  section.append(list);
+  const more = element("button", "加载更多");
+  more.type = "button";
+  more.id = "channel-more";
+  more.className = "secondary";
+  more.hidden = !state.channelCursor;
+  more.addEventListener("click", () => {
+    if (parent) {
+      loadChannel(alias, parent).catch((error) => {
+        if (error && error.message === "SESSION_EXPIRED") expireSession();
+      });
+    }
+  });
+  section.append(more);
+  if (SAFE_ID.test(alias) && SAFE_ID.test(parent)) {
+    section.append(renderChannelCompose(alias, parent, members));
+  }
   section.append(element("p", "产物尚未开放"));
   return section;
 }
 
+function reloadOpenChannel() {
+  const alias = state.detailAlias;
+  const parent = state.familyParent;
+  if (!alias || !parent) return;
+  resetChannelState();
+  loadChannel(alias, parent).catch((error) => {
+    if (error && error.message === "SESSION_EXPIRED") expireSession();
+  });
+}
+
+async function loadChannel(alias, parent) {
+  if (!hasSurface("families") || document.hidden || !SAFE_ID.test(alias) || !SAFE_ID.test(parent)) {
+    return;
+  }
+  const filter = channelFilterText();
+  const params = new URLSearchParams();
+  if (state.channelCursor) params.set("after", state.channelCursor);
+  if (filter) params.set("filter", filter);
+  const query = params.toString() ? `?${params.toString()}` : "";
+  const key = `channel:${alias}:${parent}:${state.channelCursor || "0"}:${filter || "all"}`;
+  const payload = await request(
+    `/api/v1/workspaces/${encodeURIComponent(alias)}/families/${encodeURIComponent(parent)}/channel${query}`,
+    key,
+  );
+  if (!payload || !payload.data) return;
+  state.channelMembers = Array.isArray(payload.data.members) ? payload.data.members.map((item) => text(item)).filter(Boolean) : [];
+  appendChannelMessages(Array.isArray(payload.data.messages) ? payload.data.messages : []);
+  const cursor = text(payload.data.next_cursor);
+  const received = Array.isArray(payload.data.messages) ? payload.data.messages.length : 0;
+  state.channelCursor = received ? cursor : "";
+  const more = $("channel-more");
+  if (more) more.hidden = !state.channelCursor || !received;
+  const from = $("channel-filter-from");
+  if (from) {
+    fillSelect(
+      from,
+      [["", "全部发送者"], ...state.channelMembers.map((id) => [id, id])],
+      state.channelFrom,
+    );
+  }
+  const composeTo = $("channel-post-to");
+  if (composeTo) {
+    fillSelect(
+      composeTo,
+      [["", "全家族"], ...state.channelMembers.filter((id) => id !== "operator").map((id) => [id, id])],
+      text(composeTo.value),
+    );
+  }
+  renderChannelMessages();
+}
+
+async function postHumanChannel(alias, parent, payload) {
+  if (!SAFE_ID.test(alias) || !SAFE_ID.test(parent)) return;
+  await requestWrite(
+    `/api/v1/workspaces/${encodeURIComponent(alias)}/families/${encodeURIComponent(parent)}/channel`,
+    payload,
+  );
+  resetChannelState();
+  await loadChannel(alias, parent);
+  await refreshFamilyUnread(alias, parent);
+}
+
+async function refreshFamilyUnread(alias, parent) {
+  if (!hasSurface("families") || !SAFE_ID.test(alias) || !SAFE_ID.test(parent)) return;
+  try {
+    const payload = await request(
+      `/api/v1/workspaces/${encodeURIComponent(alias)}/families/${encodeURIComponent(parent)}`,
+      `family:${alias}:${parent}`,
+    );
+    const nodes = payload && payload.data && Array.isArray(payload.data.nodes) ? payload.data.nodes : [];
+    const tree = $("family-tree");
+    if (!tree) return;
+    for (const node of nodes) {
+      const unread = tree.querySelector(`[data-member="${text(node.id)}"] .family-unread`);
+      if (unread) unread.textContent = `未读 ${count(node.unread)}`;
+    }
+  } catch (error) {
+    if (error && error.message === "SESSION_EXPIRED") throw error;
+  }
+}
+
 function renderLivePanes(alias, data) {
   const root = element("div");
   root.className = "live-panes";
@@ -1144,7 +1507,7 @@ function renderLivePanes(alias, data) {
   root.append(tabs);
   const lines = Array.isArray(data && data.lines) ? data.lines : [];
   const tasks = Array.isArray(data && data.tasks) ? data.tasks : [];
-  root.append(renderFamilyTree(alias, lines, tasks), renderEventPane(), renderChannelPane());
+  root.append(renderFamilyTree(alias, lines, tasks), renderEventPane(), renderChannelPane(alias));
   return root;
 }
 
@@ -1154,12 +1517,15 @@ function resetEventState() {
   state.eventItems = [];
   state.liveEdges = new Set();
   state.sseFailed = false;
+  resetChannelState();
 }
 
 async function loadWorkspace(alias, silent = false) {
   if (!SAFE_ID.test(alias)) return;
   if (state.detailAlias && state.detailAlias !== alias) {
     resetEventState();
+  } else {
+    resetChannelState();
   }
   try {
     const payload = await request(`/api/v1/workspaces/${encodeURIComponent(alias)}`, `workspace:${alias}`);
@@ -1191,9 +1557,14 @@ async function loadWorkspace(alias, silent = false) {
     content.append(await loadProofInspect(alias));
     content.append(renderLivePanes(alias, payload.data));
     renderEventList();
+    renderChannelMessages();
     detail.hidden = false;
     $("detail-heading").focus();
     await startEventLive(alias);
+    if (state.familyParent) {
+      await refreshFamilyUnread(alias, state.familyParent);
+      await loadChannel(alias, state.familyParent);
+    }
   } catch (error) {
     if (error && error.message === "SESSION_EXPIRED") {
       expireSession();
diff --git a/src/dyro/console/assets/styles.css b/src/dyro/console/assets/styles.css
index 57e225f..9f6d838 100644
--- a/src/dyro/console/assets/styles.css
+++ b/src/dyro/console/assets/styles.css
@@ -363,6 +363,35 @@ footer { border-top: 1px solid var(--border); color: var(--muted); font-size: .8
 .event-list { list-style: none; max-height: 16rem; overflow: auto; padding: 0; }
 .event-list li { border-top: 1px solid var(--border); padding: .35rem 0; }
 .event-list li:first-child { border-top: 0; }
+.channel-toolbar {
+  align-items: center;
+  display: flex;
+  flex-wrap: wrap;
+  gap: .4rem;
+  margin: .5rem 0;
+}
+.channel-views { display: flex; gap: .35rem; }
+.channel-list { list-style: none; max-height: 16rem; overflow: auto; padding: 0; }
+.channel-item { border-top: 1px solid var(--border); padding: .4rem 0; }
+.channel-item:first-child { border-top: 0; }
+.channel-item.retracted { text-decoration: line-through; }
+.channel-item.unacked { font-weight: 600; }
+.channel-compose {
+  display: grid;
+  gap: .45rem;
+  margin-top: .75rem;
+}
+.channel-compose select,
+.channel-compose textarea {
+  background: var(--surface);
+  border: 1px solid var(--border-strong);
+  border-radius: .35rem;
+  color: var(--text);
+  font: inherit;
+  padding: .35rem .45rem;
+  width: 100%;
+}
+.channel-toolbar select { max-width: 100%; }
 
 @media (max-width: 767px) {
   .live-panes { grid-template-columns: 1fr; }
diff --git a/src/dyro/console/families.py b/src/dyro/console/families.py
index 3dd0590..c598f70 100644
--- a/src/dyro/console/families.py
+++ b/src/dyro/console/families.py
@@ -1,29 +1,73 @@
-"""Console DTO for one-level family graphs. Channel POST is P2 and absent."""
+"""Console DTO for one-level family graphs and the family channel."""
 
 from __future__ import annotations
 
+import base64
 from collections.abc import Mapping, Sequence
+import hashlib
+import hmac
+import json
+from typing import Any
 
-from ..families import OPERATOR_ID, family_children, family_graph, family_ids
+from ..canonical import canonical_json_bytes
+from ..config import Config
+from ..families import (
+    CHANNEL_KINDS,
+    DEFAULT_CHANNEL_LIMIT,
+    HUMAN_POST_KINDS,
+    MAX_CHANNEL_LIMIT,
+    OPERATOR_ID,
+    OPERATOR_POST_KINDS,
+    FamilyChannelError,
+    ack_channel_message,
+    channel_at,
+    family_children,
+    family_graph,
+    family_ids,
+    family_members,
+    infer_post_family,
+    line_records,
+    post_channel_message,
+    read_acks,
+    read_visible_channel,
+    retracted_message_ids,
+    unread_by_member,
+)
+from .overview import ConsoleOverviewError
 from .redaction import REDACTED, safe_id
 
 
+_CURSOR_SCHEMA = 2
+_CURSOR_MAX_LENGTH = 512
+_DIGEST_HEX = frozenset("0123456789abcdef")
+_FILTER_KEYS = frozenset({"unacked", "kind", "from"})
+
+
 def _line_id(value: object) -> str:
     token = safe_id(value)
     return "" if token == REDACTED else token
 
 
+def _member_token(value: object) -> str:
+    if value == OPERATOR_ID:
+        return OPERATOR_ID
+    return _line_id(value)
+
+
 def family_badges(
     lines: Sequence[Mapping[str, object]],
     tasks: Sequence[Mapping[str, object]],
+    *,
+    unread: Mapping[str, int] | None = None,
 ) -> dict[str, dict[str, object]]:
-    """P1 badges: in-progress from tasks; dirty / missing-origin are uninspected."""
+    """P1 git badges stay uninspected.  Unread comes from the overlay channel."""
     in_progress: set[str] = set()
     for task in tasks:
         if task.get("status") == "in_progress":
             line_id = _line_id(task.get("line"))
             if line_id:
                 in_progress.add(line_id)
+    marks = dict(unread or {})
     badges: dict[str, dict[str, object]] = {}
     for line in lines:
         line_id = _line_id(line.get("id"))
@@ -33,13 +77,13 @@ def family_badges(
             "dirty": False,
             "missing_origin": False,
             "in_progress": line_id in in_progress,
-            "unread": 0,
+            "unread": marks.get(line_id, 0) if type(marks.get(line_id, 0)) is int else 0,
         }
     badges[OPERATOR_ID] = {
         "dirty": False,
         "missing_origin": False,
         "in_progress": False,
-        "unread": 0,
+        "unread": marks.get(OPERATOR_ID, 0) if type(marks.get(OPERATOR_ID, 0)) is int else 0,
     }
     return badges
 
@@ -47,9 +91,12 @@ def family_badges(
 def family_cards(
     lines: Sequence[Mapping[str, object]],
     tasks: Sequence[Mapping[str, object]],
+    *,
+    unread: Mapping[str, int] | None = None,
 ) -> list[dict[str, object]]:
-    badges = family_badges(lines, tasks)
+    badges = family_badges(lines, tasks, unread=unread)
     cards: list[dict[str, object]] = []
+    counts = dict(unread or {})
     for parent_id in family_ids(lines):
         safe_parent = _line_id(parent_id)
         if not safe_parent:
@@ -57,11 +104,14 @@ def family_cards(
         children = [_line_id(item) for item in family_children(lines, safe_parent)]
         children = [item for item in children if item]
         marks = [badges.get(safe_parent, {}), *(badges.get(child, {}) for child in children)]
+        unread_count = counts.get(safe_parent, 0)
+        if type(unread_count) is not int or unread_count < 0:
+            unread_count = 0
         cards.append(
             {
                 "parent": safe_parent,
                 "children": children,
-                "unread": 0,
+                "unread": unread_count,
                 "dirty": sum(1 for item in marks if item.get("dirty")),
                 "missing_origin": sum(1 for item in marks if item.get("missing_origin")),
                 "in_progress": sum(1 for item in marks if item.get("in_progress")),
@@ -74,6 +124,293 @@ def family_payload(
     lines: Sequence[Mapping[str, object]],
     parent_id: str,
     tasks: Sequence[Mapping[str, object]],
+    *,
+    unread: Mapping[str, int] | None = None,
 ) -> dict[str, object]:
-    graph = family_graph(lines, parent_id, badges=family_badges(lines, tasks))
+    graph = family_graph(lines, parent_id, badges=family_badges(lines, tasks, unread=unread))
     return dict(graph)
+
+
+def family_unread_maps(
+    config: Config,
+    lines: Sequence[Mapping[str, object]],
+) -> tuple[dict[str, int], dict[str, dict[str, int]]]:
+    """Return card unread (operator) and per-family member unread."""
+    cards: dict[str, int] = {}
+    members: dict[str, dict[str, int]] = {}
+    for parent_id in family_ids(lines):
+        safe_parent = _line_id(parent_id)
+        if not safe_parent:
+            continue
+        counts = unread_by_member(config, safe_parent, lines)
+        members[safe_parent] = counts
+        cards[safe_parent] = counts.get(OPERATOR_ID, 0)
+    return cards, members
+
+
+def channel_cursor_digest(record: Mapping[str, object]) -> str:
+    payload = {
+        "kind": record.get("kind", ""),
+        "at": record.get("at", ""),
+        "from": record.get("from", ""),
+        "to": record.get("to", ""),
+        "family": record.get("family", ""),
+        "body": record.get("body", ""),
+        "retracts": record.get("retracts", ""),
+    }
+    return hashlib.sha256(canonical_json_bytes(payload)).hexdigest()
+
+
+def encode_channel_cursor(
+    secret: bytes,
+    *,
+    after_seq: int,
+    message_id: str,
+    digest: str,
+) -> str:
+    body = canonical_json_bytes(
+        {
+            "schema_version": _CURSOR_SCHEMA,
+            "after": after_seq,
+            "event_id": message_id,
+            "digest": digest,
+        }
+    )
+    signature = hmac.new(secret, body, hashlib.sha256).digest()
+    return base64.urlsafe_b64encode(body + signature).rstrip(b"=").decode("ascii")
+
+
+def decode_channel_cursor(secret: bytes, value: str) -> tuple[int, str, str]:
+    if not isinstance(value, str) or not value or len(value) > _CURSOR_MAX_LENGTH:
+        raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID")
+    try:
+        raw = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+    except (ValueError, UnicodeError) as exc:
+        raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID") from exc
+    if len(raw) <= hashlib.sha256().digest_size:
+        raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID")
+    body, signature = raw[:-32], raw[-32:]
+    expected = hmac.new(secret, body, hashlib.sha256).digest()
+    if not hmac.compare_digest(expected, signature):
+        raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID")
+    try:
+        decoded: Any = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID") from exc
+    digest = decoded.get("digest") if isinstance(decoded, dict) else None
+    if (
+        not isinstance(decoded, dict)
+        or set(decoded) != {"schema_version", "after", "event_id", "digest"}
+        or decoded["schema_version"] != _CURSOR_SCHEMA
+        or type(decoded["after"]) is not int
+        or decoded["after"] < 1
+        or not isinstance(decoded["event_id"], str)
+        or not isinstance(digest, str)
+        or len(digest) != 64
+        or any(char not in _DIGEST_HEX for char in digest)
+    ):
+        raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID")
+    return decoded["after"], decoded["event_id"], digest
+
+
+def parse_channel_filter(value: str | None) -> dict[str, str | bool]:
+    if not value:
+        return {}
+    parsed: dict[str, str | bool] = {}
+    for token in value.split(","):
+        item = token.strip()
+        if not item:
+            raise ConsoleOverviewError("CHANNEL_FILTER_INVALID")
+        if item == "unacked":
+            parsed["unacked"] = True
+            continue
+        key, separator, raw = item.partition(":")
+        if not separator or key not in {"kind", "from"} or key in parsed or not raw:
+            raise ConsoleOverviewError("CHANNEL_FILTER_INVALID")
+        if key == "kind":
+            if raw not in CHANNEL_KINDS:
+                raise ConsoleOverviewError("CHANNEL_FILTER_INVALID")
+            parsed["kind"] = raw
+            continue
+        member = _member_token(raw)
+        if not member:
+            raise ConsoleOverviewError("CHANNEL_FILTER_INVALID")
+        parsed["from"] = member
+    if set(parsed) - _FILTER_KEYS:
+        raise ConsoleOverviewError("CHANNEL_FILTER_INVALID")
+    return parsed
+
+
+def project_channel_message(
+    record: Mapping[str, object],
+    *,
+    acked: frozenset[str],
+    retracted_ids: frozenset[str],
+) -> dict[str, object]:
+    message_id = str(record.get("id") or "")
+    kind = record.get("kind")
+    seq = record.get("seq")
+    at = record.get("at")
+    if type(seq) is not int or not isinstance(at, str) or kind not in CHANNEL_KINDS:
+        return {
+            "id": "msg_0",
+            "seq": 0,
+            "at": "",
+            "family": "",
+            "from": "",
+            "to": "",
+            "kind": "EVENT_REDACTED",
+            "body": "",
+            "retracts": "",
+            "retracted": False,
+            "acked": False,
+        }
+    sender = _member_token(record.get("from"))
+    recipient = record.get("to")
+    recipient_token = "" if recipient == "" else _member_token(recipient)
+    family = _line_id(record.get("family"))
+    body = record.get("body") if isinstance(record.get("body"), str) else ""
+    if len(body) > 2048:
+        body = ""
+    retracts = record.get("retracts") if isinstance(record.get("retracts"), str) else ""
+    return {
+        "id": message_id,
+        "seq": seq,
+        "at": at,
+        "family": family,
+        "from": sender,
+        "to": recipient_token,
+        "kind": kind,
+        "body": body,
+        "retracts": retracts,
+        "retracted": message_id in retracted_ids,
+        "acked": message_id in acked,
+    }
+
+
+def channel_page(
+    config: Config,
+    parent_id: str,
+    *,
+    secret: bytes,
+    after: str | None,
+    filter_text: str | None = None,
+    limit: int = DEFAULT_CHANNEL_LIMIT,
+) -> dict[str, object]:
+    if type(limit) is not int or not 1 <= limit <= MAX_CHANNEL_LIMIT:
+        raise ConsoleOverviewError("CHANNEL_LIMIT_INVALID")
+    lines = line_records(config)
+    if parent_id not in family_ids(lines):
+        raise ConsoleOverviewError("FAMILY_NOT_FOUND")
+    after_seq = 0
+    if after:
+        after_seq, message_id, digest = decode_channel_cursor(secret, after)
+        current = channel_at(config, parent_id, after_seq)
+        if (
+            current is None
+            or current.get("id") != message_id
+            or channel_cursor_digest(current) != digest
+        ):
+            raise ConsoleOverviewError("CHANNEL_CURSOR_INVALID")
+    filters = parse_channel_filter(filter_text)
+    try:
+        records = [
+            item
+            for item in read_visible_channel(config, parent_id, viewer=OPERATOR_ID)
+            if int(item["seq"]) > after_seq
+        ]
+        acked = read_acks(config, parent_id)
+        retracted_ids = retracted_message_ids(config, parent_id)
+    except FamilyChannelError as exc:
+        raise ConsoleOverviewError(exc.code) from exc
+    messages = [
+        project_channel_message(item, acked=acked, retracted_ids=retracted_ids)
+        for item in records
+    ]
+    if filters.get("unacked"):
+        messages = [item for item in messages if not item["acked"]]
+    kind_filter = filters.get("kind")
+    if isinstance(kind_filter, str):
+        messages = [item for item in messages if item["kind"] == kind_filter]
+    from_filter = filters.get("from")
+    if isinstance(from_filter, str):
+        messages = [item for item in messages if item["from"] == from_filter]
+    messages = messages[:limit]
+    if messages:
+        last = next(item for item in records if item.get("id") == messages[-1]["id"])
+        next_cursor = encode_channel_cursor(
+            secret,
+            after_seq=int(last["seq"]),
+            message_id=str(last["id"]),
+            digest=channel_cursor_digest(last),
+        )
+    elif after:
+        next_cursor = after
+    else:
+        next_cursor = None
+    return {
+        "family": parent_id,
+        "members": list(family_members(lines, parent_id)),
+        "messages": messages,
+        "next_cursor": next_cursor,
+    }
+
+
+def apply_human_channel_post(
+    config: Config,
+    parent_id: str,
+    payload: Mapping[str, object],
+    *,
+    clock=None,
+) -> dict[str, object]:
+    """Listener-side overlay write.  ``from`` is always ``operator``."""
+    if set(payload) - {"kind", "to", "body", "ack_id"}:
+        raise ConsoleOverviewError("FAMILY_POST_INVALID")
+    kind = payload.get("kind")
+    if kind not in HUMAN_POST_KINDS:
+        raise ConsoleOverviewError("FAMILY_POST_FORBIDDEN")
+    to_raw = payload.get("to", "")
+    body_raw = payload.get("body", "")
+    ack_id = payload.get("ack_id", "")
+    if to_raw is None:
+        to_raw = ""
+    if body_raw is None:
+        body_raw = ""
+    if ack_id is None:
+        ack_id = ""
+    if not isinstance(to_raw, str) or not isinstance(body_raw, str) or not isinstance(ack_id, str):
+        raise ConsoleOverviewError("FAMILY_POST_INVALID")
+    lines = line_records(config)
+    if parent_id not in family_ids(lines):
+        raise ConsoleOverviewError("FAMILY_NOT_FOUND")
+    members = set(family_members(lines, parent_id))
+    try:
+        if kind == "ack":
+            if body_raw or to_raw:
+                raise ConsoleOverviewError("FAMILY_POST_INVALID")
+            result = ack_channel_message(
+                config, ack_id, family=parent_id, clock=clock
+            )
+            if result["family"] != parent_id:
+                raise ConsoleOverviewError("CHANNEL_MESSAGE_NOT_FOUND")
+            return {"id": result["id"], "seq": result["seq"]}
+        if ack_id:
+            raise ConsoleOverviewError("FAMILY_POST_INVALID")
+        if kind not in OPERATOR_POST_KINDS:
+            raise ConsoleOverviewError("FAMILY_POST_FORBIDDEN")
+        recipient = to_raw
+        if recipient and recipient not in members:
+            raise ConsoleOverviewError("FAMILY_TO_INVALID")
+        infer_post_family(lines, OPERATOR_ID, recipient, parent_id)
+        result = post_channel_message(
+            config,
+            sender=OPERATOR_ID,
+            kind=kind,
+            body=body_raw,
+            recipient=recipient,
+            family=parent_id,
+            clock=clock,
+        )
+    except FamilyChannelError as exc:
+        raise ConsoleOverviewError(exc.code) from exc
+    return {"id": result["id"], "seq": result["seq"]}
diff --git a/src/dyro/console/inspection.py b/src/dyro/console/inspection.py
index ab47787..db7659f 100644
--- a/src/dyro/console/inspection.py
+++ b/src/dyro/console/inspection.py
@@ -155,6 +155,57 @@ def families(self, alias: str) -> dict[str, object]:
     def family(self, alias: str, parent: str) -> dict[str, object]:
         return self._request({"op": "family", "alias": alias, "parent": parent})
 
+    def channel(
+        self,
+        alias: str,
+        parent: str,
+        *,
+        after: str | None = None,
+        filter: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, object]:
+        request: dict[str, object] = {
+            "op": "channel",
+            "alias": alias,
+            "parent": parent,
+            "limit": limit,
+        }
+        if after:
+            request["after"] = after
+        if filter:
+            request["filter"] = filter
+        return self._request(request)
+
+    def post_channel(
+        self, alias: str, parent: str, payload: Mapping[str, object]
+    ) -> dict[str, object]:
+        """Write overlay signals in the listener. Never start the inspection worker."""
+        from ..config import load
+        from ..hub import WorkspaceRecord, WorkspaceRegistry, load_registry_from_home
+        from .overview import ConsoleOverviewService
+
+        if self._target_root is not None:
+            config = load(self._target_root)
+            registry = WorkspaceRegistry(
+                default=config.name,
+                workspaces=(WorkspaceRecord(name=config.name, root=config.root),),
+            )
+
+            def registry_loader() -> WorkspaceRegistry:
+                return registry
+
+        else:
+            home = self._registry_state_home
+
+            def registry_loader() -> WorkspaceRegistry:
+                return load_registry_from_home(home)
+
+        service = ConsoleOverviewService(
+            registry_loader=registry_loader,
+            cursor_secret=self._cursor_secret,
+        )
+        return service.post_channel(alias, parent, payload)
+
     def system(self) -> dict[str, object]:
         return self._request({"op": "system"})
 
@@ -342,6 +393,9 @@ def _validate_data(
         if expected_operation == "family":
             cls._validate_family(data)
             return
+        if expected_operation == "channel":
+            cls._validate_channel(data)
+            return
         if expected_operation == "workspace":
             if set(data) != {"workspace", "lines", "tasks", "objectives"}:
                 raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
@@ -516,6 +570,68 @@ def _validate_family(cls, data: dict[str, object]) -> None:
             ):
                 raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
 
+    @classmethod
+    def _validate_channel(cls, data: dict[str, object]) -> None:
+        if set(data) != {"family", "members", "messages", "next_cursor"}:
+            raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+        if not cls._safe_alias(data.get("family")):
+            raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+        members = data["members"]
+        messages = data["messages"]
+        if (
+            not isinstance(members, list)
+            or not isinstance(messages, list)
+            or len(members) > 1000
+            or len(messages) > 100
+        ):
+            raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+        for member in members:
+            if member != "operator" and not cls._safe_alias(member):
+                raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+        for item in messages:
+            if not isinstance(item, dict) or set(item) != {
+                "id",
+                "seq",
+                "at",
+                "family",
+                "from",
+                "to",
+                "kind",
+                "body",
+                "retracts",
+                "retracted",
+                "acked",
+            }:
+                raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+            sender = item.get("from")
+            recipient = item.get("to")
+            body = item.get("body")
+            retracts = item.get("retracts")
+            at = item.get("at")
+            kind = item.get("kind")
+            if (
+                type(item.get("seq")) is not int
+                or item["seq"] < 0
+                or not isinstance(item.get("id"), str)
+                or len(str(item.get("id"))) > 80
+                or not isinstance(at, str)
+                or len(at) > 40
+                or not (item.get("family") == "" or cls._safe_alias(item.get("family")))
+                or not (sender == "operator" or cls._safe_alias(sender))
+                or not (recipient == "" or recipient == "operator" or cls._safe_alias(recipient))
+                or not (kind == "EVENT_REDACTED" or cls._safe_code(kind))
+                or not isinstance(body, str)
+                or len(body) > 2048
+                or not isinstance(retracts, str)
+                or len(retracts) > 80
+                or type(item.get("retracted")) is not bool
+                or type(item.get("acked")) is not bool
+            ):
+                raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+        cursor = data["next_cursor"]
+        if cursor is not None and (not isinstance(cursor, str) or not _CURSOR.fullmatch(cursor)):
+            raise ConsoleOverviewError("OVERVIEW_UNAVAILABLE")
+
     @classmethod
     def _validate_system(cls, data: dict[str, object]) -> None:
         if set(data) != {"tool_inspection", "tools", "update"}:
diff --git a/src/dyro/console/overview.py b/src/dyro/console/overview.py
index 93f0110..7c355bc 100644
--- a/src/dyro/console/overview.py
+++ b/src/dyro/console/overview.py
@@ -9,7 +9,7 @@
 from __future__ import annotations
 
 import base64
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
 import hashlib
 import hmac
@@ -263,27 +263,89 @@ def events(
         return self._envelope(data, set())
 
     def families(self, alias: str) -> dict[str, object]:
-        """Return one-level family cards. Channel unread stays 0 in P1."""
-        from .families import family_cards
+        """Return one-level family cards. Unread is operator-unacked overlay."""
+        from .families import family_cards, family_unread_maps
 
+        config, warning_from_config = self._workspace_config(alias)
         _summary, warning_codes, inventory = self._workspace_inventory(alias)
-        data = {"families": family_cards(inventory["lines"], inventory["tasks"])}
+        warning_codes = set(warning_codes) | warning_from_config
+        card_unread, _members = family_unread_maps(config, inventory["lines"])
+        data = {
+            "families": family_cards(
+                inventory["lines"], inventory["tasks"], unread=card_unread
+            )
+        }
         return self._envelope(data, warning_codes)
 
     def family(self, alias: str, parent: str) -> dict[str, object]:
         """Return ``F(parent)``. Grandchildren are excluded."""
-        from .families import family_payload
+        from .families import family_payload, family_unread_maps
 
         try:
             parent_id = validate_id(parent, "父开发线 ID")
         except ValidationError:
             raise ConsoleOverviewError("FAMILY_PARENT_INVALID") from None
+        config, warning_from_config = self._workspace_config(alias)
         _summary, warning_codes, inventory = self._workspace_inventory(alias)
-        payload = family_payload(inventory["lines"], parent_id, inventory["tasks"])
+        warning_codes = set(warning_codes) | warning_from_config
+        _cards, members = family_unread_maps(config, inventory["lines"])
+        payload = family_payload(
+            inventory["lines"],
+            parent_id,
+            inventory["tasks"],
+            unread=members.get(parent_id, {}),
+        )
         if not payload:
             raise ConsoleOverviewError("FAMILY_NOT_FOUND")
         return self._envelope(payload, warning_codes)
 
+    def channel(
+        self,
+        alias: str,
+        parent: str,
+        *,
+        after: str | None = None,
+        filter: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, object]:
+        """Return a cursor page of the family channel. Overview polling never calls this."""
+        from .families import channel_page
+
+        try:
+            parent_id = validate_id(parent, "父开发线 ID")
+        except ValidationError:
+            raise ConsoleOverviewError("FAMILY_PARENT_INVALID") from None
+        config, warning_codes = self._workspace_config(alias)
+        try:
+            data = channel_page(
+                config,
+                parent_id,
+                secret=self._cursor_secret,
+                after=after,
+                filter_text=filter,
+                limit=limit,
+            )
+        except ConsoleOverviewError:
+            raise
+        return self._envelope(data, warning_codes)
+
+    def post_channel(
+        self,
+        alias: str,
+        parent: str,
+        payload: Mapping[str, object],
+    ) -> dict[str, object]:
+        """Write one operator overlay signal in the listener process. No git."""
+        from .families import apply_human_channel_post
+
+        try:
+            parent_id = validate_id(parent, "父开发线 ID")
+        except ValidationError:
+            raise ConsoleOverviewError("FAMILY_PARENT_INVALID") from None
+        config, warning_codes = self._workspace_config(alias)
+        data = apply_human_channel_post(config, parent_id, payload)
+        return self._envelope(data, warning_codes)
+
     def system(self) -> dict[str, object]:
         """Return cached update facts. Do not probe PATH or start a network check."""
         warnings: set[str] = set()
diff --git a/src/dyro/console/server.py b/src/dyro/console/server.py
index a60c033..0efef13 100644
--- a/src/dyro/console/server.py
+++ b/src/dyro/console/server.py
@@ -9,7 +9,7 @@
 import socket
 import threading
 from typing import Any
-from urllib.parse import urlsplit
+from urllib.parse import unquote, urlsplit
 
 from .. import __version__
 from .assets import ConsoleAssetError, load_asset, validate_assets
@@ -23,6 +23,7 @@
 HEADER_LIMIT = 16 * 1024
 HEADER_LINE_LIMIT = 4 * 1024
 SESSION_BODY_LIMIT = 512
+CHANNEL_BODY_LIMIT = 4 * 1024
 READ_TIMEOUT_SECONDS = 5.0
 REQUEST_DEADLINE_SECONDS = 10.0
 MAX_CONCURRENT_REQUESTS = 8
@@ -44,7 +45,9 @@ def _query_allowed(path: str) -> bool:
     if path == "/api/v1/overview":
         return True
     return path.startswith("/api/v1/workspaces/") and (
-        path.endswith("/events") or path.endswith("/events/stream")
+        path.endswith("/events")
+        or path.endswith("/events/stream")
+        or path.endswith("/channel")
     )
 
 
@@ -313,12 +316,12 @@ def _dispatch(self) -> None:
                     "data": {
                         "version": __version__,
                         "surfaces": (
-                            ["overview", "proofs", "system", "events"]
+                            ["overview", "proofs", "system", "events", "families"]
                             if self.console.overview_service
                             else []
                         ),
                         "capabilities": (
-                            ["overview", "proofs", "system", "events"]
+                            ["overview", "proofs", "system", "events", "families"]
                             if self.console.overview_service
                             else []
                         ),
@@ -351,6 +354,16 @@ def _dispatch(self) -> None:
             self._system()
             return
         if parsed.path.startswith("/api/v1/workspaces/"):
+            if self.command == "POST":
+                remainder = parsed.path.removeprefix("/api/v1/workspaces/")
+                parts = remainder.split("/")
+                if len(parts) == 4 and parts[1] == "families" and parts[3] == "channel":
+                    if self._authorized_session() is None:
+                        return
+                    self._channel_post(parsed.path)
+                    return
+                self._method_not_allowed()
+                return
             if self.command != "GET":
                 self._method_not_allowed()
                 return
@@ -417,7 +430,9 @@ def _workspace_resource(self, path: str, query: str) -> None:
             self._error(400, "WORKSPACE_ALIAS_INVALID")
             return
         suffix = parts[1:]
-        if query and suffix[:1] != ["events"]:
+        if query and suffix[:1] != ["events"] and not (
+            len(suffix) == 3 and suffix[0] == "families" and suffix[2] == "channel"
+        ):
             self._error(400, "BAD_REQUEST")
             return
         try:
@@ -439,6 +454,15 @@ def _workspace_resource(self, path: str, query: str) -> None:
                     self._error(400, "FAMILY_PARENT_INVALID")
                     return
                 payload = service.family(alias, parent)
+            elif len(suffix) == 3 and suffix[0] == "families" and suffix[2] == "channel":
+                parent = suffix[1]
+                if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,79}", parent):
+                    self._error(400, "FAMILY_PARENT_INVALID")
+                    return
+                after, filter_text, limit = self._channel_parameters(query)
+                payload = service.channel(
+                    alias, parent, after=after, filter=filter_text, limit=limit
+                )
             else:
                 self._error(404, "NOT_FOUND")
                 return
@@ -462,8 +486,24 @@ def _overview_error_status(code: str) -> int:
             "EVENT_QUERY_INVALID",
             "EVENT_LOG_INVALID",
             "FAMILY_PARENT_INVALID",
+            "CHANNEL_CURSOR_INVALID",
+            "CHANNEL_LIMIT_INVALID",
+            "CHANNEL_FILTER_INVALID",
+            "CHANNEL_QUERY_INVALID",
+            "CHANNEL_BODY_INVALID",
+            "CHANNEL_KIND_INVALID",
+            "FAMILY_POST_INVALID",
+            "FAMILY_TO_INVALID",
+            "FAMILY_REQUIRED",
+            "FAMILY_MEMBER_INVALID",
+            "CHANNEL_MESSAGE_NOT_FOUND",
+            "CHANNEL_MESSAGE_AMBIGUOUS",
+            "CHANNEL_LOG_INVALID",
+            "CHANNEL_ACKS_INVALID",
         }:
             return 400
+        if code in {"FAMILY_POST_FORBIDDEN"}:
+            return 403
         if code in {"WORKSPACE_NOT_FOUND", "FAMILY_NOT_FOUND"}:
             return 404
         return 503
@@ -521,6 +561,85 @@ def _event_parameters(query: str) -> tuple[str | None, int]:
             raise ConsoleOverviewError("EVENT_LIMIT_INVALID")
         return after, limit
 
+    @staticmethod
+    def _channel_parameters(query: str) -> tuple[str | None, str | None, int]:
+        if not query:
+            return None, None, 50
+        values: dict[str, str] = {}
+        for segment in query.split("&"):
+            key, separator, value = segment.partition("=")
+            if (
+                not separator
+                or key not in {"after", "filter", "limit"}
+                or key in values
+                or not value
+            ):
+                raise ConsoleOverviewError("CHANNEL_QUERY_INVALID")
+            values[key] = value
+        after = values.get("after")
+        if after is not None and (
+            len(after) > 512 or not re.fullmatch(r"[A-Za-z0-9_-]+", after)
+        ):
+            raise ConsoleOverviewError("CHANNEL_QUERY_INVALID")
+        filter_text = values.get("filter")
+        if filter_text is not None:
+            filter_text = unquote(filter_text)
+            if len(filter_text) > 256 or any(
+                ord(char) < 32 or ord(char) == 127 for char in filter_text
+            ):
+                raise ConsoleOverviewError("CHANNEL_QUERY_INVALID")
+        raw_limit = values.get("limit", "50")
+        if len(raw_limit) > 3 or not raw_limit.isdecimal():
+            raise ConsoleOverviewError("CHANNEL_QUERY_INVALID")
+        limit = int(raw_limit)
+        if not 1 <= limit <= 100:
+            raise ConsoleOverviewError("CHANNEL_LIMIT_INVALID")
+        return after, filter_text, limit
+
+    def _channel_post(self, path: str) -> None:
+        service = self.console.overview_service
+        if service is None:
+            self._error(404, "NOT_FOUND")
+            return
+        remainder = path.removeprefix("/api/v1/workspaces/")
+        parts = remainder.split("/")
+        if len(parts) != 4 or parts[1] != "families" or parts[3] != "channel":
+            self._method_not_allowed()
+            return
+        alias, parent = parts[0], parts[2]
+        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,79}", alias):
+            self._error(400, "WORKSPACE_ALIAS_INVALID")
+            return
+        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,79}", parent):
+            self._error(400, "FAMILY_PARENT_INVALID")
+            return
+        if not self._valid_origin(required=False):
+            self._error(403, "ORIGIN_REJECTED")
+            return
+        content_types = self.headers.get_all("Content-Type") or []
+        length = self._content_length(limit=CHANNEL_BODY_LIMIT)
+        if len(content_types) != 1 or content_types[0] != "application/json" or length is None:
+            self._error(400, "BAD_REQUEST")
+            return
+        try:
+            raw = self.rfile.read(length)
+            decoded = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError, OSError):
+            self._error(400, "BAD_REQUEST")
+            return
+        if not isinstance(decoded, dict):
+            self._error(400, "FAMILY_POST_INVALID")
+            return
+        try:
+            payload = service.post_channel(alias, parent, decoded)
+        except ConsoleOverviewError as exc:
+            self._error(self._overview_error_status(exc.code), exc.code)
+            return
+        except AttributeError:
+            self._error(405, "METHOD_NOT_ALLOWED")
+            return
+        self._json(200, payload, etag=str(payload.get("snapshot_sha256", "")))
+
     def _sse(self, payload: object) -> None:
         data = payload.get("data") if isinstance(payload, dict) else None
         events = data.get("events") if isinstance(data, dict) else None
@@ -563,12 +682,12 @@ def _validate_request_envelope(self) -> bool:
     def _has_body(self) -> bool:
         return bool(self.headers.get_all("Content-Length"))
 
-    def _content_length(self) -> int | None:
+    def _content_length(self, *, limit: int = SESSION_BODY_LIMIT) -> int | None:
         values = self.headers.get_all("Content-Length") or []
         if len(values) != 1 or not values[0].isdigit():
             return None
         value = int(values[0])
-        return value if value <= SESSION_BODY_LIMIT else None
+        return value if value <= limit else None
 
     def _valid_origin(self, *, required: bool) -> bool:
         origins = self.headers.get_all("Origin") or []
diff --git a/src/dyro/events.py b/src/dyro/events.py
index 23dc401..9076dc4 100644
--- a/src/dyro/events.py
+++ b/src/dyro/events.py
@@ -222,7 +222,17 @@ def event_at(config: Config, seq: int) -> dict[str, object] | None:
     return records[seq - 1]
 
 
-def append_event(
+def overlay_lock(config: Config):
+    """Shared overlay lock for ``events.jsonl`` and family channel writes."""
+    return exclusive_lock(config.root / EVENTS_LOCK)
+
+
+def read_event_records_locked(config: Config) -> list[dict[str, object]]:
+    """Read the event log.  Caller must already hold ``overlay_lock``."""
+    return _read_locked_records(events_path(config))
+
+
+def append_event_locked(
     config: Config,
     *,
     kind: str,
@@ -232,7 +242,7 @@ def append_event(
     facts: Mapping[str, object] | None = None,
     clock: Callable[[], datetime] | None = None,
 ) -> dict[str, object]:
-    """Append one typed event.  Dry-run callers must not invoke this."""
+    """Append one event.  Caller must already hold ``overlay_lock``."""
     if kind not in EVENT_KINDS:
         raise EventLogError("EVENT_WRITE_INVALID")
     actor = _safe_token(actor)
@@ -241,27 +251,53 @@ def append_event(
     cleaned = _clean_facts(facts)
     stamp = _utc(clock).strftime("%Y-%m-%dT%H:%M:%SZ")
     path = events_path(config)
-    lock = config.root / EVENTS_LOCK
     try:
-        with exclusive_lock(lock):
-            records = _read_locked_records(path)
-            seq = (records[-1]["seq"] + 1) if records else 1
-            record = {
-                "seq": seq,
-                "id": f"evt_{seq}",
-                "kind": kind,
-                "at": stamp,
-                "actor": actor,
-                "subject": subject,
-                "family": family,
-                "facts": cleaned,
-            }
-            append_text(
-                path,
-                json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n",
-            )
+        records = _read_locked_records(path)
+        seq = (records[-1]["seq"] + 1) if records else 1
+        record = {
+            "seq": seq,
+            "id": f"evt_{seq}",
+            "kind": kind,
+            "at": stamp,
+            "actor": actor,
+            "subject": subject,
+            "family": family,
+            "facts": cleaned,
+        }
+        append_text(
+            path,
+            json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n",
+        )
     except EventLogError:
         raise
     except OSError as exc:
         raise EventLogError("EVENT_WRITE_FAILED") from exc
     return record
+
+
+def append_event(
+    config: Config,
+    *,
+    kind: str,
+    actor: str,
+    subject: str,
+    family: str = "",
+    facts: Mapping[str, object] | None = None,
+    clock: Callable[[], datetime] | None = None,
+) -> dict[str, object]:
+    """Append one typed event.  Dry-run callers must not invoke this."""
+    try:
+        with overlay_lock(config):
+            return append_event_locked(
+                config,
+                kind=kind,
+                actor=actor,
+                subject=subject,
+                family=family,
+                facts=facts,
+                clock=clock,
+            )
+    except EventLogError:
+        raise
+    except OSError as exc:
+        raise EventLogError("EVENT_WRITE_FAILED") from exc
diff --git a/src/dyro/families.py b/src/dyro/families.py
index f9b3710..f1408ff 100644
--- a/src/dyro/families.py
+++ b/src/dyro/families.py
@@ -1,15 +1,83 @@
 """One-level line families: ``F(P) = {P} ∪ children(P) ∪ {operator}``.
 
-P1 uses this only to project a graph.  Channel, ack, and artifact stores are
-P2 / P3 and are not created here.
+P2 adds the overlay channel and operator ack index.  A channel write and its
+matching ``signal`` event share the overlay lock; one side only is fail-closed.
 """
 
 from __future__ import annotations
 
-from collections.abc import Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping
+from datetime import datetime, timezone
+import json
+import re
+import unicodedata
+from pathlib import Path
+from typing import Any
+
+from .config import Config, validate_id
+from .errors import DyroError, ValidationError
+from .events import (
+    EventLogError,
+    append_event_locked,
+    overlay_lock,
+    read_event_records_locked,
+)
+from .state import append_text, atomic_write_text
 
 
 OPERATOR_ID = "operator"
+CHANNEL_KINDS = frozenset(
+    {
+        "contract",
+        "blocked",
+        "shipped",
+        "ask_sync",
+        "decision",
+        "artifact",
+        "retract",
+    }
+)
+OPERATOR_POST_KINDS = frozenset({"decision", "contract"})
+HUMAN_POST_KINDS = frozenset({"decision", "contract", "ack"})
+UNACKED_KIND_PRIORITY = (
+    "blocked",
+    "ask_sync",
+    "contract",
+    "decision",
+    "shipped",
+    "artifact",
+    "retract",
+)
+DEFAULT_CHANNEL_LIMIT = 50
+MAX_CHANNEL_LIMIT = 100
+MAX_CHANNEL_BODY = 2048
+CHANNEL_FILE = "channel.jsonl"
+ACKS_FILE = "acks.json"
+MAX_CHANNEL_LOG_BYTES = 2 * 1024 * 1024
+_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,79}$")
+_CONTROL = re.compile(r"[\x00-\x1f\x7f-\x9f]")
+_CREDENTIAL = re.compile(
+    r"(?i)(?:"
+    r"(?:token|secret|password|api[_-]?key|authorization)\s*(?:=|:)"
+    r"|(?:token|secret|password|api[_-]?key|authorization)\s+[A-Za-z0-9._-]{8,}"
+    r"|(?:token|secret|password|api[_-]?key|authorization)[._-][A-Za-z0-9._-]{6,}"
+    r"|bearer\s+[A-Za-z0-9._-]{8,}"
+    r"|xox[abprs]-[A-Za-z0-9-]{10,}"
+    r"|(?:sk|rk|pk|ghp|gho|ghs|ghu|github_pat|glpat|npm|pypi|AIza)[_-][A-Za-z0-9._-]{6,}"
+    r"|AKIA[A-Z0-9]{16}"
+    r"|ya29\.[A-Za-z0-9._-]{8,}"
+    r")"
+)
+_REMOTE = re.compile(r"(?i)(?:[a-z][a-z0-9+.-]*://|git@[^\s:]+:)")
+_ABSOLUTE_PATH = re.compile(r"(?:^|[^A-Za-z0-9._-])(?:~|/|[A-Za-z]:[\\/])")
+
+
+class FamilyChannelError(DyroError):
+    """Stable, path-free failure while reading or writing a family channel."""
+
+    def __init__(self, code: str) -> None:
+        self.code = code
+        super().__init__(code)
 
 
 def family_children(lines: Iterable[Mapping[str, object]], parent_id: str) -> tuple[str, ...]:
@@ -38,7 +106,7 @@ def family_graph(
     *,
     badges: Mapping[str, Mapping[str, object]] | None = None,
 ) -> dict[str, object]:
-    """Return the P1 graph for ``F(parent_id)``.
+    """Return the one-level graph for ``F(parent_id)``.
 
     ``badges`` maps line id → ``dirty`` / ``missing_origin`` / ``in_progress`` /
     ``unread``.  Unknown ids default to false / zero.  ``operator`` has no git
@@ -64,6 +132,641 @@ def family_graph(
     }
 
 
+def line_records(config: Config) -> list[dict[str, str]]:
+    from .workspace import list_lines
+
+    return [{"id": line.id, "parent": line.parent} for line in list_lines(config)]
+
+
+def line_parent_map(lines: Iterable[Mapping[str, object]]) -> dict[str, str]:
+    return {
+        str(line.get("id") or ""): str(line.get("parent") or "")
+        for line in lines
+        if line.get("id")
+    }
+
+
+def family_dir(config: Config, parent_id: str) -> Path:
+    validate_id(parent_id, "父开发线 ID")
+    return config.root / ".dyro" / "families" / parent_id
+
+
+def channel_path(config: Config, parent_id: str) -> Path:
+    return family_dir(config, parent_id) / CHANNEL_FILE
+
+
+def acks_path(config: Config, parent_id: str) -> Path:
+    return family_dir(config, parent_id) / ACKS_FILE
+
+
+def sanitize_channel_body(value: object, *, allow_empty: bool = False) -> str:
+    if not isinstance(value, str):
+        raise FamilyChannelError("CHANNEL_BODY_INVALID")
+    normalized = unicodedata.normalize("NFC", value).strip()
+    if not normalized:
+        if allow_empty:
+            return ""
+        raise FamilyChannelError("CHANNEL_BODY_INVALID")
+    if (
+        len(normalized) > MAX_CHANNEL_BODY
+        or _CONTROL.search(normalized)
+        or _CREDENTIAL.search(normalized)
+        or _REMOTE.search(normalized)
+        or _ABSOLUTE_PATH.search(normalized)
+    ):
+        raise FamilyChannelError("CHANNEL_BODY_INVALID")
+    return normalized
+
+
+def visible_to(viewer: str, post: Mapping[str, object], family_parent: str) -> bool:
+    """Whether ``viewer`` may see ``post`` inside ``F(family_parent)``.
+
+    Broadcasts are visible to the whole family.  Directed posts are visible to
+    the sender, receiver, parent, and operator.  Cousins do not see others' DMs.
+    """
+    recipient = str(post.get("to") or "")
+    if not recipient:
+        return True
+    sender = str(post.get("from") or "")
+    return viewer in {sender, recipient, family_parent, OPERATOR_ID}
+
+
+def infer_post_family(
+    lines: Iterable[Mapping[str, object]],
+    sender: str,
+    recipient: str = "",
+    family: str = "",
+) -> str:
+    items = [dict(line) for line in lines]
+    ids = family_ids(items)
+    if family:
+        if family not in ids:
+            raise FamilyChannelError("FAMILY_NOT_FOUND")
+        members = set(family_members(items, family))
+        if sender not in members:
+            raise FamilyChannelError("FAMILY_MEMBER_INVALID")
+        if recipient and recipient not in members:
+            raise FamilyChannelError("FAMILY_TO_INVALID")
+        return family
+    if sender == OPERATOR_ID:
+        if recipient and recipient != OPERATOR_ID:
+            parent = line_parent_map(items).get(recipient, "")
+            chosen = parent or recipient
+            if chosen not in ids:
+                raise FamilyChannelError("FAMILY_NOT_FOUND")
+            members = set(family_members(items, chosen))
+            if recipient not in members:
+                raise FamilyChannelError("FAMILY_TO_INVALID")
+            return chosen
+        raise FamilyChannelError("FAMILY_REQUIRED")
+    if sender not in ids:
+        raise FamilyChannelError("FAMILY_MEMBER_INVALID")
+    parent = line_parent_map(items).get(sender, "")
+    # operator sits in every F(P). Inferring from membership would pick
+    # F(sender) for `--to operator`, hiding the post from the parent inbox.
+    if recipient == OPERATOR_ID or not recipient:
+        return parent or sender
+    own_lines = {sender, *family_children(items, sender)}
+    if recipient in own_lines:
+        return sender
+    if parent and recipient in set(family_members(items, parent)):
+        return parent
+    raise FamilyChannelError("FAMILY_TO_INVALID")
+
+
+def family_unacked(
+    config: Config,
+    *,
+    lines: Iterable[Mapping[str, object]] | None = None,
+) -> dict[str, object]:
+    """Operator-unacked overlay summary.  Never a repair or spawn blocker."""
+    items = list(lines) if lines is not None else line_records(config)
+    best: dict[str, object] | None = None
+    count = 0
+    for parent_id in family_ids(items):
+        try:
+            posts = read_visible_channel(config, parent_id, viewer=OPERATOR_ID)
+        except FamilyChannelError:
+            continue
+        acked = read_acks(config, parent_id)
+        for post in posts:
+            if post["id"] in acked:
+                continue
+            count += 1
+            if best is None or _kind_rank(str(post["kind"])) < _kind_rank(str(best["kind"])):
+                best = post
+    if best is None:
+        return {"count": 0, "kind": "", "family": "", "summary": ""}
+    return {
+        "count": count,
+        "kind": best["kind"],
+        "family": best["family"],
+        "summary": _safe_summary(str(best.get("body") or best["kind"])),
+    }
+
+
+def unread_by_member(
+    config: Config,
+    parent_id: str,
+    lines: Iterable[Mapping[str, object]],
+) -> dict[str, int]:
+    members = family_members(lines, parent_id)
+    try:
+        posts = read_visible_channel(config, parent_id, viewer=OPERATOR_ID)
+        acked = read_acks(config, parent_id)
+    except FamilyChannelError:
+        return {member: 0 for member in members}
+    unacked = [post for post in posts if post["id"] not in acked]
+    counts = {member: 0 for member in members}
+    for member in members:
+        counts[member] = sum(1 for post in unacked if visible_to(member, post, parent_id))
+    return counts
+
+
+def post_channel_message(
+    config: Config,
+    *,
+    sender: str,
+    kind: str,
+    body: str = "",
+    recipient: str = "",
+    family: str = "",
+    clock: Callable[[], datetime] | None = None,
+    dry_run: bool = False,
+) -> dict[str, object]:
+    """Append one channel row and its ``signal`` event under the overlay lock."""
+    sender = _member_id(sender, "发送者")
+    recipient = _member_id(recipient, "接收者", allow_empty=True)
+    if kind not in CHANNEL_KINDS:
+        raise FamilyChannelError("CHANNEL_KIND_INVALID")
+    if sender == OPERATOR_ID and kind not in OPERATOR_POST_KINDS:
+        raise FamilyChannelError("FAMILY_POST_FORBIDDEN")
+    lines = line_records(config)
+    parent_id = infer_post_family(lines, sender, recipient, family)
+    members = set(family_members(lines, parent_id))
+    if sender not in members or (recipient and recipient not in members):
+        raise FamilyChannelError("FAMILY_TO_INVALID")
+    retracts = ""
+    if kind == "retract":
+        retracts = _message_id(body)
+        body = ""
+    else:
+        body = sanitize_channel_body(body, allow_empty=kind == "artifact")
+    record = {
+        "from": sender,
+        "to": recipient,
+        "kind": kind,
+        "body": body,
+        "retracts": retracts,
+        "family": parent_id,
+    }
+    if dry_run:
+        return {**record, "id": "msg_0", "seq": 0, "at": "", "dry_run": True}
+    return _commit_channel_row(config, parent_id, record, clock=clock)
+
+
+def ack_channel_message(
+    config: Config,
+    message_id: str,
+    *,
+    family: str = "",
+    clock: Callable[[], datetime] | None = None,
+    dry_run: bool = False,
+) -> dict[str, object]:
+    """Mark one row operator-read.  Ack is inbox state, not a channel kind."""
+    message_id = _message_id(message_id)
+    located = find_channel_message(config, message_id, family=family)
+    if located is None:
+        raise FamilyChannelError("CHANNEL_MESSAGE_NOT_FOUND")
+    parent_id, row = located
+    if dry_run:
+        return {
+            "id": message_id,
+            "seq": row["seq"],
+            "family": parent_id,
+            "acked": True,
+            "dry_run": True,
+        }
+    try:
+        with overlay_lock(config):
+            _assert_channel_paired(config, parent_id)
+            acked = set(_read_ack_ids_locked(config, parent_id))
+            acked.add(message_id)
+            append_event_locked(
+                config,
+                kind="signal",
+                actor=OPERATOR_ID,
+                subject=parent_id,
+                family=parent_id,
+                facts={"channel_id": message_id, "ack": True},
+                clock=clock,
+            )
+            _write_acks_locked(config, parent_id, acked)
+            _assert_channel_paired(config, parent_id)
+    except EventLogError as exc:
+        raise FamilyChannelError(exc.code) from exc
+    except OSError as exc:
+        raise FamilyChannelError("CHANNEL_WRITE_FAILED") from exc
+    return {"id": message_id, "seq": row["seq"], "family": parent_id, "acked": True}
+
+
+def list_inbox(
+    config: Config,
+    *,
+    family: str = "",
+    viewer: str = "",
+    unacked: bool = False,
+) -> dict[str, object]:
+    lines = line_records(config)
+    parent_id, resolved_viewer = _inbox_scope(lines, family=family, viewer=viewer)
+    posts = read_visible_channel(config, parent_id, viewer=resolved_viewer)
+    acked = read_acks(config, parent_id)
+    if unacked:
+        posts = [post for post in posts if post["id"] not in acked]
+    messages = [_decorate(post, acked) for post in posts]
+    return {
+        "family": parent_id,
+        "viewer": resolved_viewer,
+        "messages": messages,
+        "unacked": sum(1 for item in messages if not item["acked"]),
+    }
+
+
+def find_channel_message(
+    config: Config,
+    message_id: str,
+    *,
+    family: str = "",
+) -> tuple[str, dict[str, object]] | None:
+    """Locate ``message_id``.  Bare ids are per-family; collisions fail closed."""
+    message_id = _message_id(message_id)
+    if family:
+        validate_id(family, "父开发线 ID")
+    ids = family_ids(line_records(config))
+    if family:
+        if family not in ids:
+            raise FamilyChannelError("FAMILY_NOT_FOUND")
+        parents = (family,)
+    else:
+        parents = ids
+    matches: list[tuple[str, dict[str, object]]] = []
+    for parent_id in parents:
+        try:
+            with overlay_lock(config):
+                records = _read_channel_locked(config, parent_id)
+        except FamilyChannelError:
+            if family:
+                raise
+            continue
+        for post in records:
+            if post["id"] == message_id:
+                matches.append((parent_id, post))
+                break
+    if family:
+        return matches[0] if matches else None
+    if len(matches) > 1:
+        raise FamilyChannelError("CHANNEL_MESSAGE_AMBIGUOUS")
+    return matches[0] if matches else None
+
+
+def read_visible_channel(
+    config: Config,
+    parent_id: str,
+    *,
+    viewer: str = OPERATOR_ID,
+) -> list[dict[str, object]]:
+    """Return every visible row.  Unpaired channel/event writes fail closed."""
+    validate_id(parent_id, "父开发线 ID")
+    with overlay_lock(config):
+        _assert_channel_paired(config, parent_id)
+        records = _read_channel_locked(config, parent_id)
+    return [row for row in records if visible_to(viewer, row, parent_id)]
+
+
+def retracted_message_ids(config: Config, parent_id: str) -> frozenset[str]:
+    validate_id(parent_id, "父开发线 ID")
+    with overlay_lock(config):
+        records = _read_channel_locked(config, parent_id)
+    return frozenset(
+        str(item["retracts"])
+        for item in records
+        if item["kind"] == "retract" and item["retracts"]
+    )
+
+
+def read_channel(
+    config: Config,
+    parent_id: str,
+    *,
+    viewer: str = OPERATOR_ID,
+    after_seq: int = 0,
+    limit: int = DEFAULT_CHANNEL_LIMIT,
+) -> list[dict[str, object]]:
+    if type(after_seq) is not int or after_seq < 0:
+        raise FamilyChannelError("CHANNEL_CURSOR_INVALID")
+    if type(limit) is not int or not 1 <= limit <= MAX_CHANNEL_LIMIT:
+        raise FamilyChannelError("CHANNEL_LIMIT_INVALID")
+    selected = [
+        row
+        for row in read_visible_channel(config, parent_id, viewer=viewer)
+        if int(row["seq"]) > after_seq
+    ]
+    return selected[:limit]
+
+
+def read_acks(config: Config, parent_id: str) -> frozenset[str]:
+    validate_id(parent_id, "父开发线 ID")
+    with overlay_lock(config):
+        return _read_ack_ids_locked(config, parent_id)
+
+
+def channel_at(config: Config, parent_id: str, seq: int) -> dict[str, object] | None:
+    if type(seq) is not int or seq < 1:
+        return None
+    with overlay_lock(config):
+        records = _read_channel_locked(config, parent_id)
+    if seq > len(records):
+        return None
+    return records[seq - 1]
+
+
+def _commit_channel_row(
+    config: Config,
+    parent_id: str,
+    record: Mapping[str, object],
+    *,
+    clock: Callable[[], datetime] | None = None,
+) -> dict[str, object]:
+    stamp = _utc(clock).strftime("%Y-%m-%dT%H:%M:%SZ")
+    try:
+        with overlay_lock(config):
+            _replay_unpaired_signals(config, parent_id, clock=clock)
+            _assert_channel_paired(config, parent_id)
+            records = _read_channel_locked(config, parent_id)
+            if record["kind"] == "retract":
+                target = str(record["retracts"])
+                if not any(item["id"] == target for item in records):
+                    raise FamilyChannelError("CHANNEL_MESSAGE_NOT_FOUND")
+            seq = (int(records[-1]["seq"]) + 1) if records else 1
+            row = {
+                "id": f"msg_{seq}",
+                "seq": seq,
+                "at": stamp,
+                "family": parent_id,
+                "from": record["from"],
+                "to": record["to"],
+                "kind": record["kind"],
+                "body": record["body"],
+                "retracts": record["retracts"],
+            }
+            append_text(
+                channel_path(config, parent_id),
+                json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n",
+            )
+            append_event_locked(
+                config,
+                kind="signal",
+                actor=str(row["from"]),
+                subject=str(row["to"] or parent_id),
+                family=parent_id,
+                facts={"channel_id": row["id"]},
+                clock=clock,
+            )
+            _assert_channel_paired(config, parent_id)
+    except FamilyChannelError:
+        raise
+    except EventLogError as exc:
+        raise FamilyChannelError(exc.code) from exc
+    except OSError as exc:
+        raise FamilyChannelError("CHANNEL_WRITE_FAILED") from exc
+    return row
+
+
+def _replay_unpaired_signals(
+    config: Config,
+    parent_id: str,
+    *,
+    clock: Callable[[], datetime] | None = None,
+) -> None:
+    records = _read_channel_locked(config, parent_id)
+    paired = _signal_channel_ids(config, parent_id)
+    for row in records:
+        if row["id"] in paired:
+            continue
+        append_event_locked(
+            config,
+            kind="signal",
+            actor=str(row["from"]),
+            subject=str(row["to"] or parent_id),
+            family=parent_id,
+            facts={"channel_id": str(row["id"])},
+            clock=clock,
+        )
+
+
+def _assert_channel_paired(config: Config, parent_id: str) -> None:
+    records = _read_channel_locked(config, parent_id)
+    paired = _signal_channel_ids(config, parent_id)
+    if any(row["id"] not in paired for row in records):
+        raise FamilyChannelError("CHANNEL_LOG_INCONSISTENT")
+
+
+def _signal_channel_ids(config: Config, parent_id: str) -> set[str]:
+    """Channel ids from ``signal`` rows for ``parent_id``.  Lock required."""
+    ids: set[str] = set()
+    for record in read_event_records_locked(config):
+        if record.get("kind") != "signal":
+            continue
+        if record.get("family") != parent_id:
+            continue
+        facts = record.get("facts")
+        if not isinstance(facts, dict):
+            continue
+        channel_id = facts.get("channel_id")
+        if isinstance(channel_id, str) and channel_id:
+            ids.add(channel_id)
+    return ids
+
+
+def _read_channel_locked(config: Config, parent_id: str) -> list[dict[str, object]]:
+    path = channel_path(config, parent_id)
+    if path.is_symlink() or (path.exists() and not path.is_file()):
+        raise FamilyChannelError("CHANNEL_LOG_INVALID")
+    if not path.exists():
+        return []
+    try:
+        if path.stat().st_size > MAX_CHANNEL_LOG_BYTES:
+            raise FamilyChannelError("CHANNEL_LOG_INVALID")
+        text = path.read_text(encoding="utf-8")
+    except FamilyChannelError:
+        raise
+    except OSError as exc:
+        raise FamilyChannelError("CHANNEL_LOG_INVALID") from exc
+    if not text:
+        return []
+    if not text.endswith("\n"):
+        raise FamilyChannelError("CHANNEL_LOG_INVALID")
+    records: list[dict[str, object]] = []
+    expected = 1
+    for line in text.splitlines():
+        if not line:
+            raise FamilyChannelError("CHANNEL_LOG_INVALID")
+        record = _decode_channel(line)
+        if record["seq"] != expected or record["id"] != f"msg_{expected}":
+            raise FamilyChannelError("CHANNEL_LOG_INVALID")
+        records.append(record)
+        expected += 1
+    return records
+
+
+def _decode_channel(raw: str) -> dict[str, object]:
+    try:
+        decoded: Any = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise FamilyChannelError("CHANNEL_LOG_INVALID") from exc
+    if not isinstance(decoded, dict):
+        raise FamilyChannelError("CHANNEL_LOG_INVALID")
+    required = ("id", "seq", "at", "family", "from", "to", "kind", "body", "retracts")
+    if any(key not in decoded for key in required):
+        raise FamilyChannelError("CHANNEL_LOG_INVALID")
+    seq = decoded.get("seq")
+    kind = decoded.get("kind")
+    if (
+        type(seq) is not int
+        or seq < 1
+        or not isinstance(decoded.get("id"), str)
+        or not isinstance(decoded.get("at"), str)
+        or not isinstance(decoded.get("family"), str)
+        or not isinstance(decoded.get("from"), str)
+        or not isinstance(decoded.get("to"), str)
+        or not isinstance(kind, str)
+        or kind not in CHANNEL_KINDS
+        or not isinstance(decoded.get("body"), str)
+        or not isinstance(decoded.get("retracts"), str)
+    ):
+        raise FamilyChannelError("CHANNEL_LOG_INVALID")
+    return {
+        "id": decoded["id"],
+        "seq": seq,
+        "at": decoded["at"],
+        "family": decoded["family"],
+        "from": decoded["from"],
+        "to": decoded["to"],
+        "kind": kind,
+        "body": decoded["body"],
+        "retracts": decoded["retracts"],
+    }
+
+
+def _read_ack_ids_locked(config: Config, parent_id: str) -> frozenset[str]:
+    path = acks_path(config, parent_id)
+    if path.is_symlink() or (path.exists() and not path.is_file()):
+        raise FamilyChannelError("CHANNEL_ACKS_INVALID")
+    if not path.exists():
+        return frozenset()
+    try:
+        decoded: Any = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise FamilyChannelError("CHANNEL_ACKS_INVALID") from exc
+    if not isinstance(decoded, dict) or decoded.get("schema_version") != 1:
+        raise FamilyChannelError("CHANNEL_ACKS_INVALID")
+    ids = decoded.get("ids")
+    if not isinstance(ids, list) or not all(isinstance(item, str) for item in ids):
+        raise FamilyChannelError("CHANNEL_ACKS_INVALID")
+    return frozenset(ids)
+
+
+def _write_acks_locked(config: Config, parent_id: str, ids: Iterable[str]) -> None:
+    payload = {
+        "schema_version": 1,
+        "ids": sorted(set(ids), key=_ack_sort_key),
+    }
+    atomic_write_text(
+        acks_path(config, parent_id),
+        json.dumps(payload, ensure_ascii=False, sort_keys=True) + "\n",
+    )
+
+
+def _inbox_scope(
+    lines: Iterable[Mapping[str, object]],
+    *,
+    family: str,
+    viewer: str,
+) -> tuple[str, str]:
+    items = [dict(line) for line in lines]
+    ids = family_ids(items)
+    if family:
+        if family not in ids:
+            raise FamilyChannelError("FAMILY_NOT_FOUND")
+        resolved = viewer or OPERATOR_ID
+        if resolved != OPERATOR_ID and resolved not in set(family_members(items, family)):
+            raise FamilyChannelError("FAMILY_MEMBER_INVALID")
+        return family, resolved
+    if viewer and viewer != OPERATOR_ID:
+        if viewer not in ids:
+            raise FamilyChannelError("FAMILY_MEMBER_INVALID")
+        parent = line_parent_map(items).get(viewer, "")
+        return parent or viewer, viewer
+    if len(ids) == 1:
+        return ids[0], OPERATOR_ID
+    raise FamilyChannelError("FAMILY_REQUIRED")
+
+
+def _decorate(post: Mapping[str, object], acked: frozenset[str]) -> dict[str, object]:
+    row = dict(post)
+    row["acked"] = row.get("id") in acked
+    return row
+
+
+def _member_id(value: str, label: str, *, allow_empty: bool = False) -> str:
+    if value == "":
+        if allow_empty:
+            return ""
+        raise FamilyChannelError("FAMILY_MEMBER_INVALID")
+    if value == OPERATOR_ID:
+        return OPERATOR_ID
+    try:
+        return validate_id(value, label)
+    except ValidationError as exc:
+        raise FamilyChannelError("FAMILY_MEMBER_INVALID") from exc
+
+
+def _message_id(value: object) -> str:
+    if not isinstance(value, str) or not re.fullmatch(r"msg_[1-9][0-9]{0,7}", value):
+        raise FamilyChannelError("CHANNEL_MESSAGE_NOT_FOUND")
+    return value
+
+
+def _kind_rank(kind: str) -> int:
+    try:
+        return UNACKED_KIND_PRIORITY.index(kind)
+    except ValueError:
+        return len(UNACKED_KIND_PRIORITY)
+
+
+def _safe_summary(value: str) -> str:
+    text = value.strip()
+    if len(text) > 80:
+        text = text[:80]
+    return text
+
+
+def _ack_sort_key(value: str) -> tuple[int, str]:
+    if value.startswith("msg_"):
+        try:
+            return (int(value[4:]), value)
+        except ValueError:
+            return (10**9, value)
+    return (10**9, value)
+
+
+def _utc(clock: Callable[[], datetime] | None) -> datetime:
+    value = clock() if clock is not None else datetime.now(timezone.utc)
+    if not isinstance(value, datetime) or value.tzinfo is None:
+        raise ValidationError("家族频道时钟必须提供带时区的 datetime")
+    return value.astimezone(timezone.utc)
+
+
 def _node(
     node_id: str,
     role: str,
diff --git a/src/dyro/hub.py b/src/dyro/hub.py
index db93a49..afd44a9 100644
--- a/src/dyro/hub.py
+++ b/src/dyro/hub.py
@@ -175,7 +175,12 @@ def _registry_from_json(
 
 
 def load_registry() -> WorkspaceRegistry:
-    path = _registry_path()
+    return load_registry_from_home(registry_home())
+
+
+def load_registry_from_home(home: Path) -> WorkspaceRegistry:
+    """Load the global workspace list from an explicit Dyro home."""
+    path = Path(home) / REGISTRY_FILE
     if not path.exists() and not path.is_symlink():
         return WorkspaceRegistry()
     if path.is_symlink() or not path.is_file():
diff --git a/src/dyro/integrations/assets/dyro-control-plane/SKILL.md b/src/dyro/integrations/assets/dyro-control-plane/SKILL.md
index d48ed26..153998a 100644
--- a/src/dyro/integrations/assets/dyro-control-plane/SKILL.md
+++ b/src/dyro/integrations/assets/dyro-control-plane/SKILL.md
@@ -17,6 +17,7 @@ When the request already supplies a workspace alias, skip global discovery and u
 - Health: `dyro --workspace <alias> doctor --format json`
 - One safe next step: `dyro --workspace <alias> next --format json`. If `briefing` is present, that is the switch-tool opening. `briefing.command` is a read (`tick`, `attention`, `explain`, or `list`), not a mutation, and not a resume of another harness conversation.
 - Lines or hotfixes: `dyro --workspace <alias> line list [--kind line|hotfix] --format json`. Observe `parent` from that JSON. Do not run `line spawn`, `line merge`, or `line sync`.
+- Family unread: `dyro --workspace <alias> line inbox --unacked --format json`. Report Observed as returned. User action must not invent `line post`, `line ack`, or merge.
 - Change Sets: `dyro --workspace <alias> changeset list --format json` or `dyro --workspace <alias> changeset verify <id> --format json`
 - Installed control-plane Skill health: `dyro integration status skill --format json`
 - Installed executor Skill health: `dyro integration status executor --format json`
@@ -52,7 +53,7 @@ If the user asks for 会审, 对抗, or Go/No-Go, follow the `dyro-board` protoc
 ## Hard safety boundary
 
 - Do not run `console`; it opens a local server and may launch a browser.
-- Do not run `dispatch`, `objective apply`, Objective lifecycle mutations, `task gates`, task execution or lifecycle commands, line/hotfix/Change Set creation, `line spawn`, `line merge`, `line sync`, integration install/sync/uninstall, setup/join/bootstrap/update, `open`, or `start`.
+- Do not run `dispatch`, `objective apply`, Objective lifecycle mutations, `task gates`, task execution or lifecycle commands, line/hotfix/Change Set creation, `line spawn`, `line merge`, `line sync`, `line post`, `line ack`, integration install/sync/uninstall, setup/join/bootstrap/update, `open`, or `start`.
 - Do not merge, push, sign off, release, publish, delete, or edit project files.
 - Do not edit Dyro state files or manufacture approval/confirmation fields.
 - Do not treat a command printed by `doctor`, `next`, a plan, or an error as permission to run it.
diff --git a/src/dyro/integrations/assets/dyro-line-family/SKILL.md b/src/dyro/integrations/assets/dyro-line-family/SKILL.md
index e17b0fe..438be83 100644
--- a/src/dyro/integrations/assets/dyro-line-family/SKILL.md
+++ b/src/dyro/integrations/assets/dyro-line-family/SKILL.md
@@ -35,8 +35,10 @@ Do not run any of:
 - `task merge`, `task signoff`, `task gates`, `task review`, `task run`
 - `objective apply`, `dispatch`, `console`, push, publish
 - `line create` / hotfix or Change Set creation
+- `line post` / `line inbox` / `line ack`
 
 Do not invent `--yes` or `--push`. Do not add `--push`.
+This slash does not send family signals and must not call `line post`, `inbox`, or `ack`.
 Default is no push; `policy.allow_push` is not permission to invent it.
 Do not restore a drifted line branch. Do not add `--include-paths`.
 
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 1da7db8..4e82a9b 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -26,7 +26,7 @@
 from dyro.tasks import load_task, status, task_template
 from dyro.tooling import ToolState, load_tool_preferences
 from dyro.updates import load_update_state
-from dyro.workspace import create_line, get_line, line_repository_path
+from dyro.workspace import create_line, get_line, line_repository_path, spawn_line
 
 from .support import WorkspaceCase, publish_origin_branch
 
@@ -53,6 +53,12 @@ def test_line_family_and_host_seed_accept_trailing_dry_run(self) -> None:
         self.assertTrue(spawn.dry_run)
         sync = parser.parse_args(["line", "sync", "child", "--dry-run"])
         self.assertTrue(sync.dry_run)
+        post = parser.parse_args(
+            ["line", "post", "core", "--kind", "ask_sync", "--body", "hi", "--dry-run"]
+        )
+        self.assertTrue(post.dry_run)
+        ack = parser.parse_args(["line", "ack", "msg_1", "--dry-run"])
+        self.assertTrue(ack.dry_run)
         seed = parser.parse_args(["host", "seed", "--dry-run"])
         self.assertTrue(seed.dry_run)
         global_merge = parser.parse_args(
@@ -1978,6 +1984,141 @@ def test_daemon_once_dispatches_backlog_task(self) -> None:
         self.assertEqual(status(config, load_task(config, "TASK-ONCE")), "review")
 
 
+class FamilyChannelCliTests(WorkspaceCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.config = load(self.root)
+        create_line(self.config, line_id="core", branch="feat/core", base="main")
+        spawn_line(self.config, "core", "pay")
+
+    def _read_json(self, *argv: str) -> dict[str, object]:
+        output = StringIO()
+        with redirect_stdout(output):
+            main(["--root", str(self.root), *argv, "--format", "json"])
+        return json.loads(output.getvalue())
+
+    def test_dry_run_post_and_ack_write_nothing(self) -> None:
+        from dyro.families import channel_path
+
+        planned = self._read_json(
+            "--dry-run",
+            "line",
+            "post",
+            "core",
+            "--kind",
+            "ask_sync",
+            "--body",
+            "请同步",
+        )
+        self.assertTrue(planned["dry_run"])
+        self.assertEqual(planned["kind"], "line_post")
+        self.assertEqual(planned["channel_kind"], "ask_sync")
+        self.assertFalse(channel_path(self.config, "core").exists())
+        with self.assertRaises(SystemExit):
+            main(
+                [
+                    "--root",
+                    str(self.root),
+                    "line",
+                    "post",
+                    "core",
+                    "--kind",
+                    "ask_sync",
+                    "--body",
+                    "请同步",
+                ]
+            )
+        self.assertFalse(channel_path(self.config, "core").exists())
+        written = self._read_json(
+            "line",
+            "post",
+            "core",
+            "--kind",
+            "ask_sync",
+            "--body",
+            "请同步",
+            "--yes",
+        )
+        ack = self._read_json("line", "ack", written["id"], "--dry-run")
+        self.assertTrue(ack["dry_run"])
+        inbox = self._read_json("line", "inbox", "--family", "core", "--unacked")
+        self.assertEqual(inbox["unacked"], 1)
+        self.assertEqual(inbox["messages"][0]["id"], written["id"])
+
+    def test_next_json_reports_unacked_without_post_or_ack_commands(self) -> None:
+        self._read_json(
+            "line",
+            "post",
+            "core",
+            "--kind",
+            "blocked",
+            "--body",
+            "家族未读",
+            "--yes",
+        )
+        payload = self._read_json("next")
+        commands = json.dumps(payload.get("commands", []))
+        self.assertNotIn("line post", commands)
+        self.assertNotIn("line ack --yes", commands)
+        self.assertNotIn("line ack", commands)
+        unacked = payload["family_unacked"]
+        self.assertEqual(unacked["count"], 1)
+        self.assertEqual(unacked["kind"], "blocked")
+        self.assertEqual(unacked["family"], "core")
+        self.assertNotEqual(payload.get("state"), "repair_required")
+
+    def test_ack_without_family_fails_closed_on_colliding_msg_ids(self) -> None:
+        first = self._read_json(
+            "line",
+            "post",
+            "core",
+            "--kind",
+            "blocked",
+            "--body",
+            "父族",
+            "--yes",
+        )
+        second = self._read_json(
+            "line",
+            "post",
+            "core_pay",
+            "--kind",
+            "blocked",
+            "--to",
+            "core_pay",
+            "--body",
+            "子族",
+            "--yes",
+        )
+        self.assertEqual(first["id"], "msg_1")
+        self.assertEqual(second["id"], "msg_1")
+        self.assertEqual(first["family"], "core")
+        self.assertEqual(second["family"], "core_pay")
+        stderr = StringIO()
+        with redirect_stderr(stderr), self.assertRaises(SystemExit) as raised:
+            main(
+                [
+                    "--root",
+                    str(self.root),
+                    "line",
+                    "ack",
+                    "msg_1",
+                    "--yes",
+                    "--format",
+                    "json",
+                ]
+            )
+        self.assertEqual(raised.exception.code, 2)
+        self.assertEqual(json.loads(stderr.getvalue())["code"], "CHANNEL_MESSAGE_AMBIGUOUS")
+        scoped = self._read_json("line", "ack", "msg_1", "--family", "core_pay", "--yes")
+        self.assertEqual(scoped["family"], "core_pay")
+        self.assertEqual(scoped["id"], "msg_1")
+        inbox_pay = self._read_json("line", "inbox", "--family", "core_pay")
+        inbox_core = self._read_json("line", "inbox", "--family", "core")
+        self.assertEqual(inbox_pay["unacked"], 0)
+        self.assertEqual(inbox_core["unacked"], 1)
+
+
 class VersionTests(unittest.TestCase):
     def test_package_version_matches_pyproject(self) -> None:
         import tomllib
diff --git a/tests/test_console_assets.py b/tests/test_console_assets.py
index 7627343..f61e735 100644
--- a/tests/test_console_assets.py
+++ b/tests/test_console_assets.py
@@ -106,10 +106,15 @@ def test_shell_exposes_a_semantic_command_center(self) -> None:
         self.assertIn("function renderFamilyTree".encode(), script.body)
         self.assertIn("function startEventLive".encode(), script.body)
         self.assertIn("function resetEventState".encode(), script.body)
+        self.assertIn("function renderChannelPane".encode(), script.body)
+        self.assertIn("function requestWrite".encode(), script.body)
         self.assertIn("events/stream".encode(), script.body)
+        self.assertIn("families/".encode(), script.body)
+        self.assertIn("以 operator 身份发送".encode(), script.body)
         self.assertIn("--dry-run line spawn".encode(), script.body)
         self.assertIn("--dry-run line merge".encode(), script.body)
         self.assertIn("--dry-run line sync".encode(), script.body)
+        self.assertIn("--dry-run line post".encode(), script.body)
         self.assertNotIn(b"--yes", script.body)
         self.assertNotIn(b"--push", script.body)
         self.assertIn("尚未开放".encode(), script.body)
diff --git a/tests/test_console_inspection.py b/tests/test_console_inspection.py
index 5172c8f..9802f7b 100644
--- a/tests/test_console_inspection.py
+++ b/tests/test_console_inspection.py
@@ -115,6 +115,35 @@ def test_events_after_cursor_and_one_level_family_survive_a_new_worker(self) ->
         self.assertIn("core_pay_fix", pay["data"]["members"])
         self.assertNotIn("core", pay["data"]["members"])
 
+        empty = service.channel("demo", "core")
+        self.assertEqual(empty["data"]["family"], "core")
+        self.assertEqual(empty["data"]["messages"], [])
+        with patch.object(service, "_run_worker", side_effect=AssertionError("worker")):
+            posted = service.post_channel(
+                "demo",
+                "core",
+                {"kind": "decision", "body": "先同步 core_pay"},
+            )
+        self.assertEqual(posted["data"]["id"], "msg_1")
+        with self.assertRaises(ConsoleOverviewError) as raised:
+            service.post_channel("demo", "core", {"kind": "blocked", "body": "禁止"})
+        self.assertEqual(raised.exception.code, "FAMILY_POST_FORBIDDEN")
+        reader = IsolatedOverviewService(
+            registry_state_home=self.home,
+            timeout_seconds=5,
+            cursor_secret=b"q" * 32,
+        )
+        page = reader.channel("demo", "core")
+        self.assertEqual(page["data"]["messages"][0]["from"], "operator")
+        self.assertEqual(page["data"]["messages"][0]["kind"], "decision")
+        events = reader.events("demo")
+        self.assertTrue(
+            any(
+                item["kind"] == "signal" and item["facts"].get("channel_id") == "msg_1"
+                for item in events["data"]["events"]
+            )
+        )
+
     def test_default_workspace_budget_tolerates_process_startup_overhead(self) -> None:
         clock = [0.0]
         record = WorkspaceRecord(name="demo", root=self.root)
diff --git a/tests/test_console_server.py b/tests/test_console_server.py
index d6faccb..dcccec1 100644
--- a/tests/test_console_server.py
+++ b/tests/test_console_server.py
@@ -112,8 +112,8 @@ def test_api_requires_exact_host_authorization_and_origin(self) -> None:
         self.assertEqual(headers["Content-Type"], "application/json; charset=utf-8")
         payload = json.loads(body)
         self.assertEqual(payload["schema_version"], 1)
-        self.assertEqual(payload["data"]["surfaces"], ["overview", "proofs", "system", "events"])
-        self.assertEqual(payload["data"]["capabilities"], ["overview", "proofs", "system", "events"])
+        self.assertEqual(payload["data"]["surfaces"], ["overview", "proofs", "system", "events", "families"])
+        self.assertEqual(payload["data"]["capabilities"], ["overview", "proofs", "system", "events", "families"])
         self.assertEqual(payload["data"]["initial_workspace"], "")
         self.assertIn("session_expires_at", payload["data"])
 
@@ -308,6 +308,25 @@ def setUp(self) -> None:
                 "edges": [],
             },
         }
+        self.overview.channel.return_value = {
+            "schema_version": 1,
+            "captured_at": "2026-08-04T12:00:00+00:00",
+            "snapshot_sha256": "8" * 64,
+            "freshness": {"state": "fresh", "partial": False, "warnings": []},
+            "data": {
+                "family": "core",
+                "members": ["core", "core_pay", "operator"],
+                "messages": [],
+                "next_cursor": None,
+            },
+        }
+        self.overview.post_channel.return_value = {
+            "schema_version": 1,
+            "captured_at": "2026-08-04T12:00:00+00:00",
+            "snapshot_sha256": "7" * 64,
+            "freshness": {"state": "fresh", "partial": False, "warnings": []},
+            "data": {"id": "msg_1", "seq": 1},
+        }
         self.server = create_console_http_server(
             port=0,
             bootstrap_secret="a" * 43,
@@ -468,8 +487,9 @@ def test_events_after_query_and_sse_resume_are_authenticated(self) -> None:
         channel, _, channel_body = self._request(
             "POST", "/api/v1/workspaces/alpha/families/core/channel", headers=headers
         )
-        self.assertEqual(channel, 405)
-        self.assertEqual(json.loads(channel_body)["error"]["code"], "METHOD_NOT_ALLOWED")
+        self.assertEqual(channel, 400)
+        self.assertEqual(json.loads(channel_body)["error"]["code"], "BAD_REQUEST")
+        self.overview.post_channel.assert_not_called()
 
     def test_family_graph_is_authenticated_get_only(self) -> None:
         bearer = self._bearer()
@@ -481,6 +501,82 @@ def test_family_graph_is_authenticated_get_only(self) -> None:
         self.assertEqual(json.loads(body)["data"]["parent"], "core")
         self.overview.family.assert_called_once_with("alpha", "core")
 
+    def test_channel_get_and_post_require_host_and_bearer(self) -> None:
+        from dyro.console.overview import ConsoleOverviewError
+
+        unauthorized, _, body = self._request(
+            "GET", "/api/v1/workspaces/alpha/families/core/channel"
+        )
+        self.assertEqual(unauthorized, 401)
+        self.assertEqual(json.loads(body)["error"]["code"], "UNAUTHORIZED")
+
+        bearer = self._bearer()
+        headers = {"Authorization": f"Bearer {bearer}", "Origin": self.origin}
+        status, _, body = self._request(
+            "GET",
+            "/api/v1/workspaces/alpha/families/core/channel?filter=unacked",
+            headers=headers,
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(json.loads(body)["data"]["family"], "core")
+        self.overview.channel.assert_called_once_with(
+            "alpha", "core", after=None, filter="unacked", limit=50
+        )
+
+        localhost, _, _ = self._request(
+            "GET",
+            "/api/v1/workspaces/alpha/families/core/channel",
+            headers={"Host": "localhost", "Authorization": f"Bearer {bearer}"},
+        )
+        self.assertEqual(localhost, 400)
+
+        unauth_post, _, unauth_body = self._request(
+            "POST",
+            "/api/v1/workspaces/alpha/families/core/channel",
+            body=json.dumps({"kind": "decision", "body": "ok"}).encode("utf-8"),
+            headers={"Content-Type": "application/json", "Origin": self.origin},
+        )
+        self.assertEqual(unauth_post, 401)
+        self.assertEqual(json.loads(unauth_body)["error"]["code"], "UNAUTHORIZED")
+
+        localhost_post, _, _ = self._request(
+            "POST",
+            "/api/v1/workspaces/alpha/families/core/channel",
+            body=json.dumps({"kind": "decision", "body": "ok"}).encode("utf-8"),
+            headers={
+                "Host": "localhost",
+                "Authorization": f"Bearer {bearer}",
+                "Content-Type": "application/json",
+            },
+        )
+        self.assertEqual(localhost_post, 400)
+
+        posted, _, posted_body = self._request(
+            "POST",
+            "/api/v1/workspaces/alpha/families/core/channel",
+            body=json.dumps({"kind": "decision", "body": "ok"}).encode("utf-8"),
+            headers={**headers, "Content-Type": "application/json"},
+        )
+        self.assertEqual(posted, 200)
+        self.assertEqual(json.loads(posted_body)["data"]["id"], "msg_1")
+        self.overview.post_channel.assert_called_once_with(
+            "alpha", "core", {"kind": "decision", "body": "ok"}
+        )
+
+        self.overview.post_channel.side_effect = ConsoleOverviewError(
+            "FAMILY_POST_FORBIDDEN"
+        )
+        forbidden, _, forbidden_body = self._request(
+            "POST",
+            "/api/v1/workspaces/alpha/families/core/channel",
+            body=json.dumps({"kind": "blocked", "body": "no"}).encode("utf-8"),
+            headers={**headers, "Content-Type": "application/json"},
+        )
+        self.assertEqual(forbidden, 403)
+        self.assertEqual(
+            json.loads(forbidden_body)["error"]["code"], "FAMILY_POST_FORBIDDEN"
+        )
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_families.py b/tests/test_families.py
index d217532..91cc2d3 100644
--- a/tests/test_families.py
+++ b/tests/test_families.py
@@ -1,12 +1,28 @@
 from __future__ import annotations
 
+import json
 import unittest
 
 from dyro.config import load
-from dyro.console.families import family_cards, family_payload
+from dyro.console.families import apply_human_channel_post, family_cards, family_payload
+from dyro.console.overview import ConsoleOverviewError
 from dyro.console.read_model import workspace_envelope
-from dyro.families import family_graph, family_members
+from dyro.events import read_events
+from dyro.families import (
+    FamilyChannelError,
+    ack_channel_message,
+    channel_path,
+    family_graph,
+    family_members,
+    family_unacked,
+    infer_post_family,
+    line_records,
+    post_channel_message,
+    read_acks,
+    read_visible_channel,
+)
 from dyro.observations import capture_workspace_read_snapshot
+from dyro.state import append_text
 from dyro.workspace import create_line, spawn_line
 
 from .support import WorkspaceCase
@@ -65,5 +81,266 @@ def test_family_cards_and_payload_use_direct_children_only(self) -> None:
         self.assertFalse(any(node["id"] == "core_pay_fix" for node in payload["nodes"]))
 
 
+class FamilyChannelTests(WorkspaceCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.config = load(self.root)
+        create_line(self.config, line_id="core", branch="feat/core", base="main")
+        spawn_line(self.config, "core", "pay")
+        spawn_line(self.config, "core", "shop")
+        spawn_line(self.config, "core_pay", "fix")
+
+    def test_cousins_see_broadcast_but_not_each_others_directed_posts(self) -> None:
+        broadcast = post_channel_message(
+            self.config,
+            sender="core_pay",
+            kind="ask_sync",
+            body="请同步父线",
+        )
+        directed = post_channel_message(
+            self.config,
+            sender="core_pay",
+            kind="blocked",
+            body="只要父线看见",
+            recipient="core",
+        )
+        self.assertEqual(broadcast["family"], "core")
+        self.assertEqual(directed["family"], "core")
+        shop = {item["id"] for item in read_visible_channel(self.config, "core", viewer="core_shop")}
+        pay = {item["id"] for item in read_visible_channel(self.config, "core", viewer="core_pay")}
+        parent = {item["id"] for item in read_visible_channel(self.config, "core", viewer="core")}
+        operator = {
+            item["id"] for item in read_visible_channel(self.config, "core", viewer="operator")
+        }
+        self.assertIn(broadcast["id"], shop)
+        self.assertNotIn(directed["id"], shop)
+        self.assertIn(directed["id"], pay)
+        self.assertIn(directed["id"], parent)
+        self.assertIn(directed["id"], operator)
+        events, _last = read_events(self.config)
+        signals = [item for item in events if item["kind"] == "signal"]
+        self.assertEqual(
+            [item["facts"]["channel_id"] for item in signals],
+            [broadcast["id"], directed["id"]],
+        )
+
+    def test_to_operator_uses_broadcast_default_family(self) -> None:
+        lines = line_records(self.config)
+        broadcast_family = infer_post_family(lines, "core_pay", "")
+        self.assertEqual(infer_post_family(lines, "core_pay", "operator"), broadcast_family)
+        self.assertEqual(broadcast_family, "core")
+        self.assertEqual(infer_post_family(lines, "core_pay_fix", "operator"), "core_pay")
+        self.assertEqual(
+            infer_post_family(lines, "core_pay_fix", "operator"),
+            infer_post_family(lines, "core_pay_fix", ""),
+        )
+
+        broadcast = post_channel_message(
+            self.config,
+            sender="core_pay",
+            kind="blocked",
+            body="默认家族",
+        )
+        to_operator = post_channel_message(
+            self.config,
+            sender="core_pay",
+            kind="blocked",
+            body="发给人类",
+            recipient="operator",
+        )
+        self.assertEqual(to_operator["family"], broadcast["family"])
+        self.assertEqual(to_operator["family"], "core")
+        parent = {item["id"] for item in read_visible_channel(self.config, "core", viewer="core")}
+        cousin = {
+            item["id"] for item in read_visible_channel(self.config, "core", viewer="core_shop")
+        }
+        self.assertIn(to_operator["id"], parent)
+        self.assertNotIn(to_operator["id"], cousin)
+
+        grandchild = post_channel_message(
+            self.config,
+            sender="core_pay_fix",
+            kind="blocked",
+            body="发给人类",
+            recipient="operator",
+        )
+        self.assertEqual(grandchild["family"], "core_pay")
+        self.assertEqual(broadcast["id"], "msg_1")
+        self.assertEqual(grandchild["id"], "msg_1")
+        core_msg1 = next(
+            item
+            for item in read_visible_channel(self.config, "core")
+            if item["id"] == "msg_1"
+        )
+        pay_msg1 = next(
+            item
+            for item in read_visible_channel(self.config, "core_pay")
+            if item["id"] == "msg_1"
+        )
+        self.assertEqual(core_msg1["family"], "core")
+        self.assertEqual(pay_msg1["family"], "core_pay")
+        self.assertEqual(core_msg1["from"], "core_pay")
+        self.assertEqual(pay_msg1["from"], "core_pay_fix")
+        self.assertEqual(pay_msg1["id"], core_msg1["id"])
+
+    def test_to_outside_family_is_rejected(self) -> None:
+        with self.assertRaises(FamilyChannelError) as raised:
+            post_channel_message(
+                self.config,
+                sender="core",
+                kind="blocked",
+                body="孙线不在本家族",
+                recipient="core_pay_fix",
+            )
+        self.assertEqual(raised.exception.code, "FAMILY_TO_INVALID")
+
+    def test_grandchild_channel_stays_off_the_grandparent_family(self) -> None:
+        child = post_channel_message(
+            self.config,
+            sender="core_pay",
+            kind="ask_sync",
+            body="请看修复线",
+            recipient="core_pay_fix",
+        )
+        self.assertEqual(child["family"], "core_pay")
+        core_ids = {item["id"] for item in read_visible_channel(self.config, "core")}
+        pay_ids = {item["id"] for item in read_visible_channel(self.config, "core_pay")}
+        self.assertNotIn(child["id"], core_ids)
+        self.assertIn(child["id"], pay_ids)
+
+    def test_colliding_msg_ids_are_family_scoped(self) -> None:
+        core_row = post_channel_message(
+            self.config,
+            sender="core_pay",
+            kind="blocked",
+            body="父族广播",
+        )
+        pay_row = post_channel_message(
+            self.config,
+            sender="core_pay_fix",
+            kind="blocked",
+            body="发给人类",
+            recipient="operator",
+        )
+        self.assertEqual(core_row["id"], "msg_1")
+        self.assertEqual(pay_row["id"], "msg_1")
+        self.assertEqual(core_row["family"], "core")
+        self.assertEqual(pay_row["family"], "core_pay")
+
+        with self.assertRaises(FamilyChannelError) as ambiguous:
+            ack_channel_message(self.config, "msg_1")
+        self.assertEqual(ambiguous.exception.code, "CHANNEL_MESSAGE_AMBIGUOUS")
+
+        http_ack = apply_human_channel_post(
+            self.config, "core_pay", {"kind": "ack", "ack_id": "msg_1"}
+        )
+        self.assertEqual(http_ack["id"], "msg_1")
+        self.assertEqual(read_acks(self.config, "core_pay"), frozenset({"msg_1"}))
+        self.assertEqual(read_acks(self.config, "core"), frozenset())
+
+        scoped = ack_channel_message(self.config, "msg_1", family="core")
+        self.assertEqual(scoped["family"], "core")
+        self.assertEqual(read_acks(self.config, "core"), frozenset({"msg_1"}))
+        self.assertEqual(read_acks(self.config, "core_pay"), frozenset({"msg_1"}))
+
+        with self.assertRaises(ConsoleOverviewError) as wrong_family:
+            apply_human_channel_post(
+                self.config, "core_shop", {"kind": "ack", "ack_id": "msg_1"}
+            )
+        self.assertEqual(wrong_family.exception.code, "CHANNEL_MESSAGE_NOT_FOUND")
+
+        append_text(
+            channel_path(self.config, "core_shop"),
+            json.dumps(
+                {
+                    "id": "msg_1",
+                    "seq": 1,
+                    "at": "2026-08-20T12:00:00Z",
+                    "family": "core_shop",
+                    "from": "core_shop",
+                    "to": "",
+                    "kind": "ask_sync",
+                    "body": "半写入",
+                    "retracts": "",
+                },
+                ensure_ascii=False,
+                sort_keys=True,
+            )
+            + "\n",
+        )
+        with self.assertRaises(FamilyChannelError) as unpaired:
+            read_visible_channel(self.config, "core_shop", viewer="core_shop")
+        self.assertEqual(unpaired.exception.code, "CHANNEL_LOG_INCONSISTENT")
+        still_core = {
+            item["id"] for item in read_visible_channel(self.config, "core")
+        }
+        self.assertIn("msg_1", still_core)
+
+    def test_unpaired_channel_row_fails_closed_and_is_not_broadcast(self) -> None:
+        append_text(
+            channel_path(self.config, "core"),
+            json.dumps(
+                {
+                    "id": "msg_1",
+                    "seq": 1,
+                    "at": "2026-08-20T12:00:00Z",
+                    "family": "core",
+                    "from": "core_pay",
+                    "to": "",
+                    "kind": "ask_sync",
+                    "body": "半写入",
+                    "retracts": "",
+                },
+                ensure_ascii=False,
+                sort_keys=True,
+            )
+            + "\n",
+        )
+        with self.assertRaises(FamilyChannelError) as raised:
+            read_visible_channel(self.config, "core", viewer="core_shop")
+        self.assertEqual(raised.exception.code, "CHANNEL_LOG_INCONSISTENT")
+        events, _last = read_events(self.config)
+        self.assertFalse(any(item["kind"] == "signal" for item in events))
+
+    def test_operator_post_rejects_non_human_kinds(self) -> None:
+        with self.assertRaises(FamilyChannelError) as raised:
+            post_channel_message(
+                self.config,
+                sender="operator",
+                kind="blocked",
+                body="人类不能发阻塞",
+                family="core",
+            )
+        self.assertEqual(raised.exception.code, "FAMILY_POST_FORBIDDEN")
+        with self.assertRaises(ConsoleOverviewError) as http:
+            apply_human_channel_post(
+                self.config, "core", {"kind": "shipped", "body": "不能发"}
+            )
+        self.assertEqual(http.exception.code, "FAMILY_POST_FORBIDDEN")
+
+    def test_dry_run_post_and_ack_write_nothing(self) -> None:
+        planned = post_channel_message(
+            self.config,
+            sender="core",
+            kind="decision",
+            body="先看一眼",
+            dry_run=True,
+        )
+        self.assertTrue(planned["dry_run"])
+        self.assertFalse(channel_path(self.config, "core").exists())
+        written = post_channel_message(
+            self.config,
+            sender="core",
+            kind="decision",
+            body="先看一眼",
+        )
+        ack = ack_channel_message(self.config, written["id"], dry_run=True)
+        self.assertTrue(ack["dry_run"])
+        unread = family_unacked(self.config)
+        self.assertEqual(unread["count"], 1)
+        self.assertEqual(unread["kind"], "decision")
+        self.assertEqual(unread["family"], "core")
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_integrations.py b/tests/test_integrations.py
index 2e73a8e..629ae2d 100644
--- a/tests/test_integrations.py
+++ b/tests/test_integrations.py
@@ -128,6 +128,7 @@ def test_packaged_skill_is_concise_and_has_required_metadata(self) -> None:
             "objective attention <id> --format json",
             "objective explain <id> --format json",
             "objective plan <id> --format json",
+            "line inbox --unacked --format json",
         ):
             self.assertIn(command, content)
         for forbidden_action in (
@@ -137,6 +138,8 @@ def test_packaged_skill_is_concise_and_has_required_metadata(self) -> None:
             "`line spawn`",
             "`line merge`",
             "`line sync`",
+            "`line post`",
+            "`line ack`",
         ):
             self.assertIn(forbidden_action, content)
         self.assertNotIn("`image`", content)
@@ -218,6 +221,12 @@ def test_packaged_line_family_skill_is_preflight_only(self) -> None:
         self.assertIn("wrong upstream", content)
         self.assertIn("missing worktree", content)
         self.assertIn("the real gate", content)
+        self.assertNotIn("line post <", content)
+        self.assertNotIn("line ack --yes", content)
+        self.assertIn("`line post`", content)
+        self.assertIn("`line inbox`", content)
+        self.assertIn("`line ack`", content)
+        self.assertIn("must not call `line post`", content)
         self.assertIn("$dyro-line-family", metadata.read_text(encoding="utf-8"))
         for line in metadata.read_text(encoding="utf-8").splitlines():
             if ": " in line: