feat: weixin login impl + sys-call reply support (v1.4.0)

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/channel/weixin.js

@@ -1,50 +1,478 @@
 'use strict';
 
 /**
- * channel.weixin — WeChat login via box-side WeChat SDK.
+ * channel.weixin — WeChat login via ilinkai.weixin.qq.com API.
+ *
+ * Ported from the reference weixin-login.js script.
+ * Requires Node.js >= 18 (global fetch).
  *
  * method: login
- *   params : { callId, timeout, emit }
+ *   params : { callId, timeout, emit, onReplyRef }
- *   returns: abort function (called on cancel)
+ *   returns: { abort, onReply }
  *
- * emit(msg) sends a sys-call reply back to VPS:
+ * emit(payload) sends a sys-call reply upstream:
- *   action:'event',    event:'qrcode',  data:{ url, expire, index }
+ *   { action:'event',    event:'qrcode',           data:{ url, expire:30, index } }
- *   action:'progress', event:'scanned', data:{ status:'waiting_confirm' }
+ *   { action:'event',    event:'need_verifycode',   data:{ retry } }
- *   action:'finish',   event:'success', data:{ wxid }
+ *   { action:'progress', event:'scanned',           data:{ status:'waiting_confirm' } }
- *   action:'finish',   event:'failed',  code, message
+ *   { action:'finish',   event:'success',           data:{ accountId } }
+ *   { action:'finish',   event:'failed',            code, message }
  *
- * TODO: integrate with a concrete WeChat SDK (wechaty / itchat / custom binary).
+ * To provide a verify code after 'need_verifycode', the upstream dispatcher
+ * should call the returned onReply({ data: { code: '1234' } }).
  */
 
-const log = require('../logger');
+const crypto = require('crypto');
+const fs     = require('fs');
+const path   = require('path');
+const os     = require('os');
+const log    = require('../logger');
 
-function login({ callId, timeout = 180, emit }) {
+// ── Constants (from reference script) ────────────────────────────────────────
-  log.info('weixin', `login requested callId=${callId} timeout=${timeout}`);
 
-  // TODO: start WeChat SDK, get QR code, watch for scan / confirm / expire
+const FIXED_BASE_URL           = 'https://ilinkai.weixin.qq.com';
-  // Example skeleton:
+const DEFAULT_BOT_TYPE         = '3';
-  //
+const QR_LONG_POLL_TIMEOUT_MS  = 35_000;
-  // const bot = startWechatyBot();
+const ACTIVE_LOGIN_TTL_MS      = 5 * 60_000;
-  //
+const MAX_QR_REFRESH_COUNT     = 3;
-  // bot.on('scan', (url, status) => {
+const CHANNEL_VERSION          = '2.4.3';
-  //   emit({ action: 'event', event: 'qrcode', data: { url, expire: 30, index: ++qrIndex } });
+const ILINK_APP_ID             = 'bot';
-  // });
+const ILINK_APP_CLIENT_VERSION = String(_buildClientVersion(CHANNEL_VERSION));
-  // bot.on('login', (user) => {
-  //   emit({ action: 'finish', event: 'success', data: { wxid: user.id } });
-  // });
-  // bot.on('error', (err) => {
-  //   emit({ action: 'finish', event: 'failed', code: 1001, message: err.message });
-  // });
-  // bot.start();
-  //
-  // return () => bot.stop();   // ← abort function
 
-  // Temporary stub: immediately report not implemented
+function _buildClientVersion(version) {
-  emit({ action: 'finish', event: 'failed', code: 501, message: 'weixin SDK not implemented' });
+  const parts = String(version).split('.').map(p => parseInt(p, 10));
+  const major = parts[0] || 0;
+  const minor = parts[1] || 0;
+  const patch = parts[2] || 0;
+  return ((major & 0xff) << 16) | ((minor & 0xff) << 8) | (patch & 0xff);
+}
 
-  return () => {
+// ── State-dir helpers (mirrors reference script) ─────────────────────────────
-    log.info('weixin', `login cancelled callId=${callId}`);
+
+function _resolveStateDir() {
+  return (
+    (process.env.OPENCLAW_STATE_DIR || '').trim() ||
+    (process.env.CLAWDBOT_STATE_DIR || '').trim() ||
+    path.join(os.homedir(), '.openclaw')
+  );
+}
+
+function _resolveWeixinStateDir()  { return path.join(_resolveStateDir(), 'openclaw-weixin'); }
+function _resolveAccountIndexPath(){ return path.join(_resolveWeixinStateDir(), 'accounts.json'); }
+function _resolveAccountsDir()     { return path.join(_resolveWeixinStateDir(), 'accounts'); }
+function _resolveAccountPath(id)   { return path.join(_resolveAccountsDir(), `${id}.json`); }
+
+const VALID_ID_RE    = /^[a-z0-9][a-z0-9_-]{0,63}$/i;
+const INVALID_CHARS  = /[^a-z0-9_-]+/g;
+
+function _normalizeAccountId(value) {
+  const trimmed = String(value || '').trim();
+  if (!trimmed) return 'default';
+  const lower = trimmed.toLowerCase();
+  if (VALID_ID_RE.test(trimmed)) return lower;
+  return lower.replace(INVALID_CHARS, '-').replace(/^-+/, '').replace(/-+$/, '').slice(0, 64) || 'default';
+}
+
+function _listIndexedAccountIds() {
+  try {
+    const p = _resolveAccountIndexPath();
+    if (!fs.existsSync(p)) return [];
+    const parsed = JSON.parse(fs.readFileSync(p, 'utf8'));
+    return Array.isArray(parsed) ? parsed.filter(id => typeof id === 'string' && id.trim()) : [];
+  } catch (_) { return []; }
+}
+
+function _registerAccountId(accountId) {
+  fs.mkdirSync(_resolveWeixinStateDir(), { recursive: true });
+  const existing = _listIndexedAccountIds();
+  if (existing.includes(accountId)) return;
+  fs.writeFileSync(_resolveAccountIndexPath(), JSON.stringify([...existing, accountId], null, 2), 'utf8');
+}
+
+function _loadAccount(accountId) {
+  try {
+    const p = _resolveAccountPath(accountId);
+    if (fs.existsSync(p)) return JSON.parse(fs.readFileSync(p, 'utf8'));
+  } catch (_) {}
+  // legacy token fallback
+  try {
+    const legacyPath = path.join(_resolveStateDir(), 'credentials', 'openclaw-weixin', 'credentials.json');
+    if (fs.existsSync(legacyPath)) {
+      const parsed = JSON.parse(fs.readFileSync(legacyPath, 'utf8'));
+      if (typeof parsed.token === 'string') return { token: parsed.token };
+    }
+  } catch (_) {}
+  return null;
+}
+
+function _saveAccount(accountId, update) {
+  fs.mkdirSync(_resolveAccountsDir(), { recursive: true });
+  const existing = _loadAccount(accountId) || {};
+  const token   = (update.token   || '').trim() || existing.token;
+  const baseUrl = (update.baseUrl || '').trim() || existing.baseUrl;
+  const userId  = (update.userId  || '').trim() || (existing.userId || '').trim() || undefined;
+  const data = {
+    ...(token   ? { token, savedAt: new Date().toISOString() } : {}),
+    ...(baseUrl ? { baseUrl } : {}),
+    ...(userId  ? { userId  } : {}),
+  };
+  const filePath = _resolveAccountPath(accountId);
+  fs.writeFileSync(filePath, JSON.stringify(data, null, 2), 'utf8');
+  try { fs.chmodSync(filePath, 0o600); } catch (_) {}
+  return filePath;
+}
+
+function _clearStaleAccountsForUserId(currentAccountId, userId) {
+  if (!userId) return;
+  for (const id of _listIndexedAccountIds()) {
+    if (id === currentAccountId) continue;
+    const data = _loadAccount(id);
+    if (data && (data.userId || '').trim() === userId) {
+      log.info('weixin', `removing stale account with same userId: ${id}`);
+      try { fs.unlinkSync(_resolveAccountPath(id)); } catch (_) {}
+      const existing = _listIndexedAccountIds();
+      fs.writeFileSync(_resolveAccountIndexPath(), JSON.stringify(existing.filter(x => x !== id), null, 2), 'utf8');
+    }
+  }
+}
+
+function _getLocalBotTokenList() {
+  const tokens = [];
+  const ids = _listIndexedAccountIds();
+  for (let i = ids.length - 1; i >= 0 && tokens.length < 10; i--) {
+    const token = (_loadAccount(ids[i]) || {}).token;
+    if (token && token.trim()) tokens.push(token.trim());
+  }
+  return tokens;
+}
+
+// ── HTTP helpers ──────────────────────────────────────────────────────────────
+
+function _commonHeaders() {
+  return {
+    'iLink-App-Id':            ILINK_APP_ID,
+    'iLink-App-ClientVersion': ILINK_APP_CLIENT_VERSION,
   };
 }
 
+function _randomWechatUin() {
+  const uint32 = crypto.randomBytes(4).readUInt32BE(0);
+  return Buffer.from(String(uint32), 'utf8').toString('base64');
+}
+
+function _postHeaders() {
+  return {
+    'Content-Type':      'application/json',
+    'AuthorizationType': 'ilink_bot_token',
+    'X-WECHAT-UIN':      _randomWechatUin(),
+    ..._commonHeaders(),
+  };
+}
+
+async function _postJson(baseUrl, endpoint, body, timeoutMs) {
+  const url = `${baseUrl.replace(/\/$/, '')}/${endpoint}`;
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), timeoutMs || 15_000);
+  try {
+    const res = await fetch(url, {
+      method: 'POST',
+      headers: _postHeaders(),
+      body: JSON.stringify(body),
+      signal: controller.signal,
+    });
+    const text = await res.text();
+    if (!res.ok) throw new Error(`HTTP ${res.status}: ${text.slice(0, 300)}`);
+    return JSON.parse(text);
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+async function _getJson(baseUrl, endpoint, timeoutMs) {
+  const url = `${baseUrl.replace(/\/$/, '')}/${endpoint}`;
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), timeoutMs || QR_LONG_POLL_TIMEOUT_MS + 5_000);
+  try {
+    const res = await fetch(url, {
+      method: 'GET',
+      headers: _commonHeaders(),
+      signal: controller.signal,
+    });
+    const text = await res.text();
+    if (!res.ok) throw new Error(`HTTP ${res.status}: ${text.slice(0, 300)}`);
+    return JSON.parse(text);
+  } catch (err) {
+    if (err && err.name === 'AbortError') return { status: 'wait' };
+    throw err;
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+// ── WeChat API calls ──────────────────────────────────────────────────────────
+
+async function _fetchQRCode(botType) {
+  const localTokenList = _getLocalBotTokenList();
+  const data = await _postJson(
+    FIXED_BASE_URL,
+    `ilink/bot/get_bot_qrcode?bot_type=${encodeURIComponent(botType)}`,
+    { local_token_list: localTokenList },
+    15_000
+  );
+  if (data.ret !== undefined && data.ret !== 0) {
+    throw new Error(`get_bot_qrcode ret=${data.ret} errmsg=${data.errmsg || ''}`);
+  }
+  if (!data.qrcode || !data.qrcode_img_content) {
+    throw new Error(`get_bot_qrcode response missing qrcode/qrcode_img_content`);
+  }
+  return data;
+}
+
+async function _pollQRStatus(apiBaseUrl, qrcode, verifyCode) {
+  try {
+    let endpoint = `ilink/bot/get_qrcode_status?qrcode=${encodeURIComponent(qrcode)}`;
+    if (verifyCode) endpoint += `&verify_code=${encodeURIComponent(verifyCode)}`;
+    const data = await _getJson(apiBaseUrl, endpoint, QR_LONG_POLL_TIMEOUT_MS);
+    if (data.ret !== undefined && data.ret !== 0) {
+      throw new Error(`get_qrcode_status ret=${data.ret} errmsg=${data.errmsg || ''}`);
+    }
+    return data;
+  } catch (err) {
+    if (err && err.name === 'AbortError') return { status: 'wait' };
+    log.warn('weixin', `pollQRStatus error (will retry): ${err.message}`);
+    return { status: 'wait' };
+  }
+}
+
+function _bumpOpenClawConfigTimestamp() {
+  const stateDir = _resolveStateDir();
+  const candidates = [
+    process.env.OPENCLAW_CONFIG || '',
+    path.join(stateDir, 'openclaw.json'),
+  ].filter(Boolean);
+  const configPath = candidates.find(p => { try { return fs.existsSync(p); } catch (_) { return false; } });
+  if (!configPath) return;
+  try {
+    const cfg = JSON.parse(fs.readFileSync(configPath, 'utf8'));
+    if (!cfg.channels) cfg.channels = {};
+    if (!cfg.channels['openclaw-weixin']) cfg.channels['openclaw-weixin'] = {};
+    cfg.channels['openclaw-weixin'].channelConfigUpdatedAt = new Date().toISOString();
+    fs.writeFileSync(configPath, JSON.stringify(cfg, null, 2), 'utf8');
+    log.info('weixin', `bumped openclaw.json timestamp: ${configPath}`);
+  } catch (err) {
+    log.warn('weixin', `bump timestamp failed: ${err.message}`);
+  }
+}
+
+// ── login() ───────────────────────────────────────────────────────────────────
+
+/**
+ * Start WeChat QR login.
+ *
+ * @param {object} opts
+ * @param {string}   opts.callId    - sys-call id (for logging)
+ * @param {number}   opts.timeout   - overall timeout in seconds (default 180)
+ * @param {string}   opts.botType   - ilink bot type (default '3')
+ * @param {function} opts.emit      - emit(payload) send event upstream
+ * @returns {{ abort: function, onReply: function }} task handle
+ */
+function login({ callId, timeout = 180, botType = DEFAULT_BOT_TYPE, emit }) {
+  let aborted  = false;
+  let finished = false;
+
+  // Pending verify-code: resolve/reject
+  let _pendingVerifyResolve = null;
+  let _pendingVerifyReject  = null;
+
+  function _waitForVerifyCode() {
+    return new Promise((resolve, reject) => {
+      _pendingVerifyResolve = resolve;
+      _pendingVerifyReject  = reject;
+    });
+  }
+
+  function onReply(msg) {
+    const event = (msg.event || '').toString();
+    const code  = (msg.data && msg.data.code) ? String(msg.data.code).trim() : '';
+    if (event === 'verify_code' && _pendingVerifyResolve) {
+      log.info('weixin', `callId=${callId} verify_code received`);
+      const resolve = _pendingVerifyResolve;
+      _pendingVerifyResolve = null;
+      _pendingVerifyReject  = null;
+      resolve(code);
+    }
+  }
+
+  function abort() {
+    aborted = true;
+    if (_pendingVerifyReject) {
+      const reject = _pendingVerifyReject;
+      _pendingVerifyResolve = null;
+      _pendingVerifyReject  = null;
+      reject(new Error('aborted'));
+    }
+    log.info('weixin', `callId=${callId} aborted`);
+  }
+
+  const timeoutMs = Math.max(timeout * 1000, 30_000);
+
+  // Run async without blocking caller
+  _runLogin({ callId, timeoutMs, botType, emit,
+    isAborted: () => aborted,
+    waitForVerifyCode: _waitForVerifyCode,
+  })
+    .then(() => { finished = true; })
+    .catch((err) => {
+      if (finished) return;
+      finished = true;
+      log.error('weixin', `callId=${callId} login error: ${err.message}`);
+      emit({ action: 'finish', event: 'failed', code: 500, message: err.message });
+    });
+
+  return { abort, onReply };
+}
+
+async function _runLogin({ callId, timeoutMs, botType, emit, isAborted, waitForVerifyCode }) {
+  let qrRefreshCount = 1;
+  let scannedEmitted = false;
+  let activeLogin    = null;
+
+  const deadline = Date.now() + timeoutMs;
+
+  async function startOrRefreshQr() {
+    const qrData = await _fetchQRCode(botType);
+    activeLogin = {
+      qrcode:         qrData.qrcode,
+      qrcodeUrl:      qrData.qrcode_img_content,
+      startedAt:      Date.now(),
+      apiBaseUrl:     FIXED_BASE_URL,
+      pendingVerify:  null,
+    };
+    emit({
+      action: 'event',
+      event:  'qrcode',
+      data:   { url: activeLogin.qrcodeUrl, expire: 30, index: qrRefreshCount },
+    });
+    log.info('weixin', `callId=${callId} qrcode emitted index=${qrRefreshCount}`);
+  }
+
+  await startOrRefreshQr();
+
+  while (!isAborted() && Date.now() < deadline) {
+    // Refresh if local TTL exceeded
+    if (Date.now() - activeLogin.startedAt >= ACTIVE_LOGIN_TTL_MS) {
+      log.info('weixin', `callId=${callId} QR TTL expired, refreshing`);
+      qrRefreshCount++;
+      if (qrRefreshCount > MAX_QR_REFRESH_COUNT) {
+        emit({ action: 'finish', event: 'failed', code: 1003, message: 'QR expired too many times' });
+        return;
+      }
+      await startOrRefreshQr();
+      scannedEmitted = false;
+      activeLogin.pendingVerify = null;
+      continue;
+    }
+
+    const status = await _pollQRStatus(
+      activeLogin.apiBaseUrl,
+      activeLogin.qrcode,
+      activeLogin.pendingVerify
+    );
+    activeLogin.pendingVerify = null; // consumed
+
+    if (isAborted()) return;
+
+    switch (status.status) {
+      case 'wait':
+        break;
+
+      case 'scaned':
+      case 'scaned_but_redirect': {
+        if (status.redirect_host) {
+          activeLogin.apiBaseUrl = `https://${status.redirect_host}`;
+        }
+        if (!scannedEmitted) {
+          scannedEmitted = true;
+          emit({ action: 'progress', event: 'scanned', data: { status: 'waiting_confirm' } });
+          log.info('weixin', `callId=${callId} scanned`);
+        }
+        break;
+      }
+
+      case 'need_verifycode': {
+        const retry = !!activeLogin.pendingVerify;
+        emit({ action: 'event', event: 'need_verifycode', data: { retry } });
+        log.info('weixin', `callId=${callId} need_verifycode (retry=${retry})`);
+        try {
+          const code = await waitForVerifyCode();
+          if (isAborted()) return;
+          activeLogin.pendingVerify = code;
+        } catch (_) {
+          // aborted while waiting
+          return;
+        }
+        break;
+      }
+
+      case 'expired': {
+        log.info('weixin', `callId=${callId} QR expired, refreshing`);
+        qrRefreshCount++;
+        if (qrRefreshCount > MAX_QR_REFRESH_COUNT) {
+          emit({ action: 'finish', event: 'failed', code: 1001, message: 'qrcode expired' });
+          return;
+        }
+        await startOrRefreshQr();
+        scannedEmitted = false;
+        break;
+      }
+
+      case 'verify_code_blocked': {
+        log.warn('weixin', `callId=${callId} verify_code_blocked`);
+        qrRefreshCount++;
+        if (qrRefreshCount > MAX_QR_REFRESH_COUNT) {
+          emit({ action: 'finish', event: 'failed', code: 1002, message: 'verify_code blocked' });
+          return;
+        }
+        await startOrRefreshQr();
+        scannedEmitted = false;
+        break;
+      }
+
+      case 'binded_redirect':
+        log.info('weixin', `callId=${callId} already connected`);
+        emit({ action: 'finish', event: 'success', data: { accountId: 'already_connected' } });
+        return;
+
+      case 'confirmed': {
+        if (!status.ilink_bot_id) throw new Error('confirmed: missing ilink_bot_id');
+        if (!status.bot_token)    throw new Error('confirmed: missing bot_token');
+        const accountId = _normalizeAccountId(status.ilink_bot_id);
+        const filePath  = _saveAccount(accountId, {
+          token:   status.bot_token,
+          baseUrl: status.baseurl,
+          userId:  status.ilink_user_id,
+        });
+        _registerAccountId(accountId);
+        if (status.ilink_user_id) _clearStaleAccountsForUserId(accountId, status.ilink_user_id);
+        _bumpOpenClawConfigTimestamp();
+        log.info('weixin', `callId=${callId} login success accountId=${accountId} file=${filePath}`);
+        emit({ action: 'finish', event: 'success', data: { accountId } });
+        return;
+      }
+
+      default:
+        log.warn('weixin', `callId=${callId} unknown status: ${JSON.stringify(status)}`);
+        break;
+    }
+
+    // Brief pause between polls to avoid tight looping on 'wait'
+    if (status.status === 'wait') {
+      await new Promise(resolve => setTimeout(resolve, 1000));
+    }
+  }
+
+  if (!isAborted()) {
+    emit({ action: 'finish', event: 'failed', code: 1004, message: 'login timeout' });
+  }
+}
+
 module.exports = { login };

lib/sys-call.js

@@ -3,22 +3,17 @@
 /**
  * sys-call dispatcher.
  *
- * Routes incoming sys-call messages (action='request'|'cancel') from VPS
+ * Routes incoming sys-call messages from VPS to channel handlers.
- * to the correct channel handler, and wires the emit callback so that
- * handler replies are sent back over the WebSocket.
  *
- * Message envelope (shared by all sys-call messages):
+ * Supported actions (incoming from VPS):
- * {
+ *   request  — start a new task
- *   type:    'sys-call',
+ *   cancel   — abort a running task
- *   id:      '<UUID>',
+ *   reply    — forward input to a running task (e.g. verify code)
- *   api:     'channel.weixin',
+ *
- *   method:  'login',
+ * Handler interface:
- *   action:  'request' | 'reply' | 'notify' | 'progress' | 'event' | 'finish' | 'cancel',
+ *   handler[method](params) → { abort, onReply? }
- *   event:   '',
+ *     abort()          — called on cancel
- *   code:    0,
+ *     onReply(msg)     — called on reply (optional)
- *   message: '',
- *   data:    {}
- * }
  */
 
 const log = require('./logger');
@@ -28,14 +23,14 @@ const handlers = {
   'channel.weixin': require('./channel/weixin'),
 };
 
-// ── running tasks: callId → abort() ──────────────────────────────────────────
+// ── running tasks: callId → { abort, onReply? } ───────────────────────────────
 const running = new Map();
 
 /**
  * Handle an incoming sys-call message from VPS.
  *
  * @param {object}   msg  - parsed message object
- * @param {function} send - send(replyPayload) → forwards to VPS over WS;
+ * @param {function} send - send(replyPayload) → forwarded to VPS over WS
  *                          caller prepends { type:'sys-call' }
  */
 function handle(msg, send) {
@@ -48,10 +43,10 @@ function handle(msg, send) {
 
   // ── cancel ────────────────────────────────────────────────────────────────
   if (action === 'cancel') {
-    const abort = running.get(callId);
+    const task = running.get(callId);
-    if (abort) {
+    if (task) {
       log.info('sys-call', `cancel callId=${callId}`);
-      try { abort(); } catch (e) { log.warn('sys-call', `abort error: ${e.message}`); }
+      try { task.abort(); } catch (e) { log.warn('sys-call', `abort error: ${e.message}`); }
       running.delete(callId);
     } else {
       log.debug('sys-call', `cancel for unknown/finished callId=${callId}`);
@@ -59,6 +54,18 @@ function handle(msg, send) {
     return;
   }
 
+  // ── reply (e.g. verify code from frontend) ────────────────────────────────
+  if (action === 'reply') {
+    const task = running.get(callId);
+    if (task && typeof task.onReply === 'function') {
+      log.info('sys-call', `reply callId=${callId} event=${msg.event || ''}`);
+      try { task.onReply(msg); } catch (e) { log.warn('sys-call', `onReply error: ${e.message}`); }
+    } else {
+      log.debug('sys-call', `reply for unknown/no-onReply callId=${callId}`);
+    }
+    return;
+  }
+
   // ── request ───────────────────────────────────────────────────────────────
   if (action !== 'request') {
     log.warn('sys-call', `unexpected action=${action} callId=${callId}`);
@@ -76,7 +83,7 @@ function handle(msg, send) {
     return;
   }
 
-  // emit: wraps handler replies, cleans up running map on finish
+  // emit: merges envelope fields, cleans up running map on finish
   const emit = (payload) => {
     send({ id: callId, api, method, event: '', code: 0, message: '', ...payload });
     if (payload.action === 'finish') {
@@ -87,9 +94,9 @@ function handle(msg, send) {
 
   log.info('sys-call', `start api=${api} method=${method} callId=${callId}`);
 
-  let abort;
+  let task;
   try {
-    abort = handler[method]({ callId, ...(msg.data || {}), emit });
+    task = handler[method]({ callId, ...(msg.data || {}), emit });
   } catch (e) {
     log.error('sys-call', `handler threw: ${e.message}`);
     send({
@@ -100,8 +107,13 @@ function handle(msg, send) {
     return;
   }
 
-  if (typeof abort === 'function') {
+  // Normalise: handler may return a function (abort only) or { abort, onReply }
-    running.set(callId, abort);
+  if (typeof task === 'function') {
+    task = { abort: task };
+  }
+
+  if (task && typeof task.abort === 'function') {
+    running.set(callId, task);
   }
 }