[eric8810]

Refs #537. This fixes the slow-path chain loss and cancellation; the host-root call and the remaining async-control classification are still open.

• with_caller captures the invocation path when the future is created and enters it on every poll, so a filter or listener that suspends before its first poll still forwards from the caller's chain.
• Frame::Cancel removes a call from the wait graph when an incoming call or an awaiting relay is removed. A synchronous call already running keeps its mark until its reply, so a later synchronous forward still participates in cycle detection.
• unchained now also marks the thread as sending a notification; waits::forward leaves such sends out of the wait graph. events.elsewhere and the link notifications use it, so they no longer enter the table.
• A forwarded call whose chain ends at a marked call remains a hop, including a relayed function reference through request_async; a notification nobody waits for is not.
• Docs: cancellation removes unstarted or asynchronously awaited calls only; a running synchronous call keeps its mark until its reply.

Tests:

• function_references_calling_each_other in sync_cycles.rs: two Node runtimes take each other's function reference, then call them while both wait; one call is refused with SyncWaitCycle.
• with_caller_captures_the_chain_when_it_is_made: the path is kept when the future is created, before its first poll.
• (b) an_emit_listener_called_after_its_filter_answered in sync_cycles.rs: a Node emitter emits through a filter, so the dispatch suspends before any listener is called; the listener, run in Python on top of a call C waits for, calls C back and is refused with SyncWaitCycle, and the rest completes. It locks the current behaviour and does not exercise with_caller: an emit's filter and listener calls run on the bus's own task (Bus::emit_erased, crates/rutis/src/bus.rs), with an empty path and no caller, so it passes with or without the change above. The eager capture in with_caller matters for the parallel/serial/waterfall dispatch futures, events/link.rs and next_function; with_caller_captures_the_chain_when_it_is_made covers it.
• (c) a_running_synchronous_call_cancelled_keeps_its_mark (rpc/tests.rs): a synchronous call cancelled while it runs keeps its mark; its later forward that closes a cycle with another session is refused with SyncWaitCycle. Without the cancel condition the forward is sent and waits for ever.
• (d) a_cancelled_awaited_call_leaves_no_mark (rpc/tests.rs): a synchronous await of a future, cancelled through a real Frame::Cancel while it is awaited, leaves the session's mark count at 0. Without the awaiting branch the mark stays.
• a_notification_sent_unchained_is_not_in_the_graph, an_answered_call_leaves_no_mark, and the existing sync_cycles suite (4 passed).
• cargo test -p rutis-bridge --all-features --lib (139 passed), cargo clippy -p rutis-bridge --all-features --all-targets -- -D warnings.

Still open under #537: a host-originated synchronous call is recorded as an entry (no caller session id), and an async notification that does not go through unchained on a marked chain is still a hop. An emit's chain is not carried into the bus's emit task. On an application thread with no tokio runtime, an emit's reply panics (no reactor running) and run_emit drops a pending run: #567.

[eric8810]

审核结论：需要修改，暂不能合并（会让 #228 的卡死回来）。

1. 跨会话转发不再查环：rpc.rs:1482、waits.rs:174 只有桥自己以 Waiting::Sync 发出时才算 hop。但同步等待的调用方经别的运行时传来的函数引用发起调用时，是在 sync=false 的任务上用 request_async 转发的（relay.rs:142-167），本 PR 下变成 entry，跳过查环。
   复现：两个 Node 运行时各拿到对方的函数，同时同步调用。main 上 0.19s 报 SyncWaitCycle；本 PR 下两边挂到 20s 超时。这也与 fix(bridge): 等待图的入口是调用链的边界（修复误报 SyncWaitCycle），检查近线性；补 Node 监听器时机的例外说明 #535 的模型不符：CrossRuntimeSync.tla 的 RustForward（覆盖 relay.rs relay）总记 hop，只有 RustPost 记 entry。
   方向：是否为 hop 仍按调用方是否被标记；不等待的通知走 unchained（如 push_elsewhere）或显式的“不等待”标志，而不是看 Waiting。
2. 取消时标记泄漏（rpc.rs:2195）：异步转发、正在等回复的调用被取消后从 awaiting 移除，respond 不再运行，标记留到会话结束，MARKS 一直 >0。条件应为 incoming.is_some() || awaiting.is_some()。
3. 路径取得太晚（rpc.rs:559-560）：with_caller 是 async fn，current_path() 在第一次 poll 时才运行；宿主忙时排队的 emit 仍会丢链。改为普通 fn 返回 impl Future，立即取路径。
4. 没有测试：0.15 review(events): 等待图对同步/异步/控制调用的分类不准，慢路径丢链 #537 列了验收。至少加：上面的引用成环、同步处理里发出的异步通知不被拒、过滤器首次 poll 时挂起的 emit 仍能查出环、取消后再转发。
5. 文档：docs/design-multilang-runtimes-2026-10-03.md:179-184（及 .en.md）仍按“链尾是被标记的调用”定义 hop。

CI：runtimes-go 失败在 go_rows::a_runtime_that_ended_before_supervising_is_handled_once_it_starts 的 20s 超时，该测试不经过改动的代码，属不稳定/环境问题，与本 PR 无关。

[eric8810]

复审：代码方向对了，还需改说明、测试与文档。

第二个提交回到“调用方被标记即为 hop”，与 #535 的模型（RustForward 总记 hop）一致，引用成环的回归已解决：实测两个 Node 运行时各持对方函数同时同步调用，main 与本 PR 都拒绝其中一个（SyncWaitCycle），另一个完成。取消条件 incoming.is_some() || awaiting.is_some() 与 with_caller 立即取路径都对。rutis-bridge 全部测试、clippy、sync_cycles 4/4 通过。

仍需：

1. 标题与正文已不符合代码：标题说“只把同步调用算 hop”，正文说标记链上的异步调用/控制通知“现在是 entry”、send 在 forward 前算 sync——本 PR 都不再这样做。0.15 review(events): 等待图对同步/异步/控制调用的分类不准，慢路径丢链 #537 的 1.2（同步处理里发出的异步通知仍可能被拒）与 1.3（events.elsewhere 仍进入等待图）仍未做。请改写标题与正文，在 “Refs 0.15 review(events): 等待图对同步/异步/控制调用的分类不准，慢路径丢链 #537” 下列出仍未做的项；或按之前的建议实现（不等待的通知走 unchained 或显式的不等待标志）。
2. 测试仍缺（唯一的新测试 an_asynchronous_forward_on_a_marked_chain_is_a_hop 测的是 main 已有行为）：
   • (a) 引用成环，放进 crates/rutis-loader/tests/sync_cycles.rs（补丁见下）；
   • (b) 过滤器首次 poll 时挂起的 emit，其监听器再同步回调成环，证明 with_caller（rpc.rs:559-577）立即取路径有效；
   • (c) 取消一个正在运行的同步调用，再从它转发成环，仍被拒（rpc.rs:2197-2202）；
   • (d) 取消一个被异步等待的调用后，等待图的标记数回到 0（覆盖泄漏修复）。
3. 文档：docs/design-multilang-runtimes-2026-10-03.md:184 与 .en.md:184 仍说取消会移除记录；现在只对未开始或被异步等待的调用如此，正在运行的同步调用在回复前保留标记。两种语言都改。

注意 (a) 的写法：函数要在更早的一次调用里取到。如果在 go 里取，main 与本 PR 都会挂住，因为宿主发起的同步调用不被记录——这正是 #537 里本 PR 延后的那一项。

另：an_emit_listener_run_while_a_call_waits 里 rpc.rs:324（Executor::current）panic “there is no reactor running”，emit 记为失败但测试仍通过（main 上也一样），已另开 https://github.com/arcships/rutis/issues/567。

(a) 的测试补丁

diff --git a/crates/rutis-loader/tests/sync_cycles.rs b/crates/rutis-loader/tests/sync_cycles.rs
index c9dd2ebf..16c70417 100644
--- a/crates/rutis-loader/tests/sync_cycles.rs
+++ b/crates/rutis-loader/tests/sync_cycles.rs
@@ -939,3 +939,86 @@ async fn an_emit_listener_run_while_a_call_waits() {
         fixture.root.shutdown().await.unwrap();
     }
 }
+
+// REVIEW-ONLY: function references relayed between two Node runtimes.
+const NODE_REF_WORK: &str = r#"
+import { definePlugin } from 'PLUGIN'
+export default definePlugin({
+  provides: { SELF: { fn: 'sync' } },
+  apply(ctx) {
+    ctx.provide('SELF', { fn: () => () => 'SELF works' })
+  },
+})
+"#;
+const NODE_REF_GO: &str = r#"
+import { definePlugin } from 'PLUGIN'
+export default definePlugin({
+  inject: ['gate', 'OTHER'],
+  provides: { SELF: { prepare: 'sync', go: 'sync' } },
+  apply(ctx) {
+    const gate = ctx.use('gate')
+    let f
+    ctx.provide('SELF', {
+      prepare() { f = ctx.use('OTHER').fn(); return 'ready' },
+      go() {
+        gate.arrive('SELF')
+        try { return `SELF got ${f()}` } catch (error) { return `SELF: ${error.name}` }
+      },
+    })
+  },
+})
+"#;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn review_function_references_calling_each_other() {
+    let dir = tempfile::tempdir().unwrap();
+    let work = |me: &str| NODE_REF_WORK.replace("SELF", me);
+    let go = |me: &str, other: &str| NODE_REF_GO.replace("SELF", me).replace("OTHER", other);
+    let files = [
+        ("aw", "node", write(dir.path(), "aw.mjs", &work("a_work"))),
+        ("bw", "nb", write(dir.path(), "bw.mjs", &work("b_work"))),
+        ("a", "node", write(dir.path(), "a.mjs", &go("a_svc", "b_work"))),
+        ("b", "nb", write(dir.path(), "b.mjs", &go("b_svc", "a_work"))),
+    ];
+    let (gate, _arrivals) = Gate::new(2);
+    let fixture = fixture(
+        vec![(node("node"), true), (node("nb"), false)],
+        false,
+        false,
+        &["a_work", "b_work", "a_svc", "b_svc"],
+        gate,
+    )
+    .await;
+    let rows_of = files
+        .iter()
+        .map(|(id, runtime, file)| match *runtime {
+            "node" => json!({ "id": id, "name": file.to_string_lossy() }),
+            runtime => json!({ "id": id, "name": format!("{runtime}:{}", file.display()) }),
+        })
+        .collect();
+    fixture.loader.reconcile(rows(rows_of), None).await.unwrap();
+    let (loader, root) = (fixture.loader.clone(), fixture.root.clone());
+    COLD.until(
+        || {
+            ["aw", "bw", "a", "b"].iter().all(|id| active(&loader, id))
+                && ["a_work", "b_work", "a_svc", "b_svc"]
+                    .iter()
+                    .all(|name| service(&root, name).is_some())
+        },
+        "the rows and their services",
+    )
+    .await;
+    answer(&call(&fixture.root, "a_svc", "prepare"), "a.prepare");
+    answer(&call(&fixture.root, "b_svc", "prepare"), "b.prepare");
+    let started = std::time::Instant::now();
+    let a_go = call(&fixture.root, "a_svc", "go");
+    let b_go = call(&fixture.root, "b_svc", "go");
+    let answers = [answer(&a_go, "a.go"), answer(&b_go, "b.go")];
+    eprintln!("REVIEW answers {answers:?} in {:?}", started.elapsed());
+    let refused = answers
+        .iter()
+        .filter(|a| a.as_str().unwrap().ends_with(": SyncWaitCycle"))
+        .count();
+    assert_eq!(refused, 1, "{answers:?}");
+    fixture.root.shutdown().await.unwrap();
+}