From fbc867f59149dcb98c736d9adfcbcbb58c96bca3 Mon Sep 17 00:00:00 2001 From: speak-agent <248744407+speak-agent@users.noreply.github.com> Date: Sun, 27 Sep 2026 22:36:33 +0800 Subject: [PATCH 1/2] feat(interface): progress for update_packages, no stray stdout in interface mode `interface update_packages` reported no progress at all for an index refresh, and when the refresh rebuilt a stale index cache it let a downloaded pkgindex-build.lua's raw terminal text (carriage returns, ANSI clear-to-end-of-line) straight onto stdout, breaking the NDJSON stream the protocol requires there. The text comes from libxpkg's plain-Lua build sandbox: run_pkgindex_build executes the index repo's own pkgindex-build.lua with the real io library, and that script's self-refreshing progress line writes directly to the process's real stdout, bypassing every abstraction xlings routes through platform::is_tui_mode(). It is data the index repo ships, not code this repository owns, so the fix is at the boundary rather than in that script. platform::StdoutCapture redirects the real fd 1 into a pipe for the duration of a capability's execution and delivers whatever is written there line by line; interface::run() installs it around every capability, converting a recognisable "[i/n] ..." step into a progress event and routing anything else to stderr. The NDJSON writer itself is decoupled from fd 1 (InterfaceSession now writes through a duplicate taken before any capture starts), so the protocol's own output is never captured along with what this exists to catch. Because the guard sits around capability execution rather than around one call site, install_packages benefits the same way when its own first-run index rebuild hits the same script. cmd_update also reports two new progress phases -- index_sync (one event per repository, i/n within its group) and index_rebuild (from the captured scan) -- and reuses install_packages' own download_progress data event shape for an index artifact download's byte progress, so a client renders both the same way. Adds interface_update_packages_progress_test.sh (E2E-122): builds an offline index fixture whose own pkgindex-build.lua reproduces the exact defect shape, asserts every stdout line parses as JSON, and asserts at least one progress event fires. Confirmed it fails against the released 2026.9.27.1 binary before this change and passes after. --- docs/spec/interface-ndjson-v1.md | 45 ++++- modules/platform/src/platform.cpp | 122 ++++++++++++ modules/platform/src/platform.cppm | 47 +++++ src/core/xim/commands.cpp | 45 ++++- src/core/xim/indexfetch.cpp | 33 +++- src/core/xim/indexfetch.cppm | 15 +- src/core/xim/repo.cpp | 74 ++++--- src/core/xim/repo.cppm | 17 +- src/interface.cpp | 90 ++++++++- src/interface.cppm | 6 + ...interface_update_packages_progress_test.sh | 183 ++++++++++++++++++ tests/e2e/run_all.sh | 5 + 12 files changed, 641 insertions(+), 41 deletions(-) create mode 100755 tests/e2e/interface_update_packages_progress_test.sh diff --git a/docs/spec/interface-ndjson-v1.md b/docs/spec/interface-ndjson-v1.md index 8fde847a2..0d2acd97e 100644 --- a/docs/spec/interface-ndjson-v1.md +++ b/docs/spec/interface-ndjson-v1.md @@ -122,7 +122,13 @@ xlings interface install_packages --args '{"targets":["gcc@14"],"yes":true}' ## 5. stdout 事件格式 -每行为一个 JSON 对象，必含 `"kind"` 字段标识类型。 +每行为一个 JSON 对象，必含 `"kind"` 字段标识类型。stdout 上不会出现这一行 JSON +之外的任何字节——包括某能力在执行期间于进程内运行的、xlings 自身不控制的代码 +（例如索引缓存失效后，通过库的 Lua 沙箱执行下载来的索引构建脚本）。这类代码 +即使直接写系统级 stdout，也会被截获：可识别为进度的输出转换成一条 `progress` +事件，其余转发到 stderr，两种情况都不出现在 stdout 上——此前 `update_packages`／ +`install_packages` 触发的索引重建会把这类脚本的原始终端文本（含 `\r`、ANSI +清行序列）直接混进 NDJSON 流，客户端解析到该行即失败。 ## 6. 事件类型 @@ -140,6 +146,21 @@ xlings interface install_packages --args '{"targets":["gcc@14"],"yes":true}' | percent | number | 0-100 整数，-1 表示不确定 | | message | string | 人类可读描述 | +#### 6.1.1 `update_packages` 的 progress 事件 + +`update_packages` 在刷新索引（未指定 `target`，或指定 `target` 时刷新之前）的 +过程中，按下列 phase 报告进度；三者都可能一次都不出现——本地/离线源没有网络 +字节可报，缓存未失效时不会重建： + +| phase | 何时出现 | i/n 的含义 | +|-------|---------|-----------| +| `index_sync` | 每同步一个配置的索引仓库 | 该仓库在其所属分组（顶层 index_repos，或一批已发现的子索引）中的序号/该分组总数 | +| `index_rebuild` | 索引缓存失效、重新解析 pkgs/ 时 | 已处理/待处理的包文件总数 | + +`percent` 为 `i/n`（分组总数为 0 时为 0）；`message` 同时给出仓库名或包文件名， +便于在没有专门 UI 的客户端里也能打印出有意义的一行。索引工件下载的字节级进度 +走 `download_progress`（§6.3.2），不走这里。 + ### 6.2 log 日志消息。 @@ -230,6 +251,28 @@ xlings interface install_packages --args '{"targets":["gcc@14"],"yes":true}' {"kind":"data","dataKind":"install_targets","payload":{"targets":[{"request":"xim:nosuchpkg","namespace":"","name":"","version":"","revision":0,"status":"failed","payload_dir":""}]}} ``` +#### 6.3.2 download_progress + +字节级下载进度，`install_packages` 与 `update_packages` 共用同一形状——后者 +用它报告索引工件（当索引仓库以工件方式获取、且字节数已知时；git/本地源没有 +这一层进度）的下载，客户端因此无需为两者各写一套渲染逻辑。 + +```json +{"kind":"data","dataKind":"download_progress","payload":{"files":[{"name":"xim","totalBytes":204800,"downloadedBytes":102400,"started":true,"finished":false,"success":false}],"nameWidth":3,"elapsedSec":1.2,"sizesReady":true,"prevLines":0}} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| files | array | 本次渲染帧里的每一个下载任务 | +| files[].name | string | 任务标签——`update_packages` 用索引仓库名，`install_packages` 用文件名 | +| files[].totalBytes | number | 总字节数；`sizesReady` 为 false 时无意义 | +| files[].downloadedBytes | number | 已下载字节数 | +| files[].started / finished / success | boolean | 该任务的生命周期 | +| nameWidth | number | 渲染对齐用的标签列宽；结构化客户端可忽略 | +| elapsedSec | number | 自本次刷新/安装开始的已用秒数 | +| sizesReady | boolean | 总字节数是否已知（服务器在响应头给出前不知道） | +| prevLines | number | 上一帧渲染的终端行数；结构化客户端可忽略 | + ### 6.4 prompt 服务端向客户端请求用户输入。 diff --git a/modules/platform/src/platform.cpp b/modules/platform/src/platform.cpp index 3b2d0bb8f..4d3529621 100644 --- a/modules/platform/src/platform.cpp +++ b/modules/platform/src/platform.cpp @@ -6,8 +6,10 @@ module; #include #include #include +#include #else #include +#include #define NOMINMAX #include #endif @@ -163,6 +165,126 @@ bool is_tui_mode() { return tui_mode_.load(std::memory_order_relaxed); } +int dup_stdout_fd() { +#if defined(_WIN32) + return ::_dup(1); +#else + return ::dup(1); +#endif + } + +bool write_fd(int fd, std::string_view data) { + if (fd < 0) return false; + const char* p = data.data(); + std::size_t remaining = data.size(); + while (remaining > 0) { +#if defined(_WIN32) + unsigned int chunk = static_cast( + std::min(remaining, 1u << 20)); + int n = ::_write(fd, p, chunk); + if (n <= 0) return false; +#else + ssize_t n = ::write(fd, p, remaining); + if (n < 0) { + if (errno == EINTR) continue; + return false; + } + if (n == 0) return false; +#endif + p += n; + remaining -= static_cast(n); + } + return true; + } + +void close_fd(int fd) { + if (fd < 0) return; +#if defined(_WIN32) + ::_close(fd); +#else + ::close(fd); +#endif + } + +StdoutCapture::StdoutCapture(std::function onLine) { + std::fflush(stdout); + savedFd_ = dup_stdout_fd(); + if (savedFd_ < 0) return; // best-effort: leave fd 1 untouched + + int fds[2]; +#if defined(_WIN32) + if (::_pipe(fds, 1u << 16, _O_BINARY) != 0) { + close_fd(savedFd_); + savedFd_ = -1; + return; + } + if (::_dup2(fds[1], 1) != 0) { + ::_close(fds[0]); + ::_close(fds[1]); + close_fd(savedFd_); + savedFd_ = -1; + return; + } + ::_close(fds[1]); +#else + if (::pipe(fds) != 0) { + close_fd(savedFd_); + savedFd_ = -1; + return; + } + if (::dup2(fds[1], 1) < 0) { + ::close(fds[0]); + ::close(fds[1]); + close_fd(savedFd_); + savedFd_ = -1; + return; + } + ::close(fds[1]); +#endif + pipeRead_ = fds[0]; + + // EOF on pipeRead_ (a zero-length read) is the signal that fd 1 was + // handed back to the real terminal in the destructor -- that closes + // the pipe's last write-end reference, which is what unblocks this + // read() without any cooperative cancellation. + reader_ = std::thread([this, onLine = std::move(onLine)]() { + std::string buf; + char chunk[4096]; + for (;;) { +#if defined(_WIN32) + int n = ::_read(pipeRead_, chunk, sizeof(chunk)); + if (n <= 0) break; +#else + ssize_t n = ::read(pipeRead_, chunk, sizeof(chunk)); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) break; +#endif + for (int i = 0; i < n; ++i) { + char c = chunk[i]; + if (c == '\n' || c == '\r') { + if (!buf.empty()) { onLine(buf); buf.clear(); } + } else { + buf.push_back(c); + } + } + } + if (!buf.empty()) onLine(buf); + }); + } + +StdoutCapture::~StdoutCapture() { + if (savedFd_ < 0) return; + std::fflush(stdout); +#if defined(_WIN32) + ::_dup2(savedFd_, 1); +#else + ::dup2(savedFd_, 1); +#endif + close_fd(savedFd_); + if (reader_.joinable()) reader_.join(); + if (pipeRead_ >= 0) close_fd(pipeRead_); + } + int exec(const std::string& cmd) { std::string actual_cmd = cmd; if (tui_mode_.load(std::memory_order_relaxed)) { diff --git a/modules/platform/src/platform.cppm b/modules/platform/src/platform.cppm index 8b1fac8ce..54c94e013 100644 --- a/modules/platform/src/platform.cppm +++ b/modules/platform/src/platform.cppm @@ -139,6 +139,53 @@ namespace platform { export bool is_tui_mode(); + // ── Real stdout, decoupled from whatever fd 1 currently means ────── + // + // The interface protocol (xlings.interface) writes NDJSON to fd 1 and + // forbids anything else there. In-process code this program does not + // control -- a Lua build script run through a vendored package-index + // loader, for the one measured case -- can still put raw text straight + // onto the C library's `stdout`, which is fd 1 underneath regardless of + // who wrote it. `StdoutCapture` below redirects fd 1 away from the real + // terminal for as long as such code might run; a writer that must keep + // reaching the real terminal throughout (the NDJSON writer itself) has + // to hold ITS OWN duplicate of fd 1, taken with dup_stdout_fd() BEFORE + // any capture starts, so the redirection cannot swallow the very output + // it exists to protect. + + // A duplicate of the process's current fd 1, usable with write_fd() / + // close_fd() regardless of what fd 1 is later redirected to. -1 on + // failure (descriptor limit, or an unsupported platform). + export int dup_stdout_fd(); + + // Write `data` in full to `fd`, retrying across short writes and EINTR. + // Returns false on a hard I/O error (fd closed, broken pipe, ...). + export bool write_fd(int fd, std::string_view data); + + export void close_fd(int fd); + + // RAII: while alive, redirects the real fd 1 into an internal pipe and + // delivers whatever is written there to `onLine`, one line at a time -- + // split on '\n' AND on a bare '\r', since a self-refreshing terminal + // progress line never sends '\n' until it is done. Restores the + // original fd 1 on destruction and joins the reader thread. + // + // Mechanical only: it does not parse what it captures. Not reentrant -- + // at most one instance should be alive per process, since a nested one's + // destructor would hand fd 1 back to the outer capture's pipe rather than + // to the real terminal. + export class StdoutCapture { + public: + explicit StdoutCapture(std::function onLine); + ~StdoutCapture(); + StdoutCapture(const StdoutCapture&) = delete; + StdoutCapture& operator=(const StdoutCapture&) = delete; + private: + int savedFd_ { -1 }; + int pipeRead_ { -1 }; + std::thread reader_; + }; + export int exec(const std::string& cmd); export [[nodiscard]] std::string shell_quote(const std::string& arg); diff --git a/src/core/xim/commands.cpp b/src/core/xim/commands.cpp index 97083d513..8f5ce66da 100644 --- a/src/core/xim/commands.cpp +++ b/src/core/xim/commands.cpp @@ -2708,8 +2708,49 @@ int cmd_clear_xpkg(const std::string& what) { } int cmd_update(const std::string& target, bool yes, EventStream& stream) { - // Sync repos - if (!sync_all_repos(true)) { + // Sync repos. Reported per repository so an interface client can show + // progress for a step that otherwise runs silently until it either + // finishes or times out (#599's XLINGS_UPDATE_TIMEOUT budget) -- there + // used to be nothing on the wire for this at all, only the heartbeat. + auto onRepoSync = [&stream](std::string_view repoName, int index, int total) { + stream.emit(ProgressEvent{ + .phase = "index_sync", + .percent = total > 0 + ? static_cast(index) / static_cast(total) : 0.0f, + .message = std::format("syncing index repo {}/{}: {}", + index, total, repoName), + }); + }; + // Bytes-level progress for whichever repo is fetched as an artifact + // (a git clone or a local-source link never calls this). Shaped exactly + // like install_packages' own `download_progress` (see DownloadProgressRenderer + // above in this file) so a client renders an index download the same way + // it already renders a package download, rather than needing a second + // code path for one more `files[]` shape. + const auto downloadStart = std::chrono::steady_clock::now(); + auto onIndexBytes = [&stream, downloadStart](std::string_view label, + double total, double downloaded) { + const bool sizesReady = total > 0.0; + const bool finished = sizesReady && downloaded >= total; + nlohmann::json files = nlohmann::json::array(); + files.push_back({ + {"name", std::string(label)}, + {"totalBytes", total}, + {"downloadedBytes", downloaded}, + {"started", true}, + {"finished", finished}, + {"success", finished} + }); + nlohmann::json payload; + payload["files"] = std::move(files); + payload["nameWidth"] = static_cast(label.size()); + payload["elapsedSec"] = std::chrono::duration( + std::chrono::steady_clock::now() - downloadStart).count(); + payload["sizesReady"] = sizesReady; + payload["prevLines"] = 0; + stream.emit(DataEvent{"download_progress", payload.dump()}); + }; + if (!sync_all_repos(true, onRepoSync, onIndexBytes)) { log::error("failed to sync repositories"); return 1; } diff --git a/src/core/xim/indexfetch.cpp b/src/core/xim/indexfetch.cpp index 230f22eff..80b61f4b4 100644 --- a/src/core/xim/indexfetch.cpp +++ b/src/core/xim/indexfetch.cpp @@ -40,7 +40,8 @@ std::string lower_hex_(std::string s) { // if its bytes match. Returns empty string on success, else the last error. std::string download_candidates_(std::vector urls, const std::filesystem::path& destFile, - std::string_view wantSha256) { + std::string_view wantSha256, + BytesProgress onProgress = nullptr) { if (urls.empty()) return "no candidate URLs"; // With a pinned sha256, mirrors may race ahead of the authoritative URL; // without one (the manifest pointer), keep the authoritative server first. @@ -92,6 +93,12 @@ std::string download_candidates_(std::vector urls, url, digest ? *digest : "", want); }; } + if (onProgress) { + opts.onProgress = [onProgress, label = destFile.filename().string()] + (double total, double now) { + onProgress(label, total, now); + }; + } auto r = tinyhttps::download_file(opts); if (r.success) return {}; lastErr = r.error.empty() ? ("failed: " + u) : r.error; @@ -138,7 +145,8 @@ BaseOverride resolve_base_() { // custom source passes an EMPTY override ("use the URL list as-is"). std::string obtain_file(const std::string& filename, std::vector remoteUrls, const std::filesystem::path& dest, std::string_view wantSha, - const BaseOverride* forced = nullptr) { + const BaseOverride* forced = nullptr, + BytesProgress onProgress = nullptr) { namespace fs = std::filesystem; auto b = forced ? *forced : resolve_base_(); @@ -165,14 +173,15 @@ std::string obtain_file(const std::string& filename, std::vector re for (const auto& e : b.entries) { const std::string& name = e.filename.empty() ? filename : e.filename; auto err = e.local ? copy_local_(*e.local, name) - : download_candidates_({ e.base + "/" + name }, dest, wantSha); + : download_candidates_({ e.base + "/" + name }, dest, wantSha, + onProgress); if (err.empty()) return {}; lastErr = err; } if (!b.entries.empty() && !b.allowRemoteUrls) return lastErr; if (remoteUrls.empty()) return lastErr.empty() ? std::string("no candidate URLs") : lastErr; - auto err = download_candidates_(std::move(remoteUrls), dest, wantSha); + auto err = download_candidates_(std::move(remoteUrls), dest, wantSha, onProgress); if (err.empty()) return {}; return lastErr.empty() ? err : lastErr + "; " + err; } @@ -624,7 +633,8 @@ bool fetch_index_artifact(const std::filesystem::path& destIndexDir, std::string& err, std::string_view subName, const ArtifactSource* custom, - std::string_view pin) { + std::string_view pin, + BytesProgress onProgress) { namespace fs = std::filesystem; auto mirrorKey = Config::mirror(); std::string key = custom ? custom->key @@ -688,10 +698,21 @@ bool fetch_index_artifact(const std::filesystem::path& destIndexDir, detail_::BaseOverride forcedStorage; const detail_::BaseOverride* forced = nullptr; if (custom) { forcedStorage = detail_::base_override_for_(*custom); forced = &forcedStorage; } + // Relabels every progress callback with the index's own name (`key`) + // rather than the artifact's filename -- that is what a client already + // reports the sync step under, and a download rendered as a second, + // differently-labelled entry for the SAME index would look like an + // unrelated one. + BytesProgress relabelled = onProgress + ? BytesProgress([onProgress, key](std::string_view, double total, double now) { + onProgress(key, total, now); + }) + : BytesProgress{}; if (auto e = detail_::obtain_file(snapshot.artifact_name, index_asset_urls(snapshot.artifact_name, mirrorKey, snapshot.index_version, custom), - artifactFile, snapshot.artifact_sha256, forced); !e.empty()) { + artifactFile, snapshot.artifact_sha256, forced, + relabelled); !e.empty()) { err = std::format("fetch index artifact failed: {}", e); return false; } diff --git a/src/core/xim/indexfetch.cppm b/src/core/xim/indexfetch.cppm index 4fa64193d..6c7f70aa9 100644 --- a/src/core/xim/indexfetch.cppm +++ b/src/core/xim/indexfetch.cppm @@ -177,6 +177,15 @@ std::vector index_pointer_urls(std::string_view filename, const std::map& load_index_pointers(std::string_view mirror, const ArtifactSource* custom = nullptr); +// Bytes-level progress for a network fetch this layer performs, labelled by +// the caller (a repo name, an artifact filename, ...) so a client can show +// what is downloading. Only the artifact fetch reports through it: a git +// clone has no byte-granular signal to give, and the pointer file ahead of +// it is too small to be worth one. +using BytesProgress = std::function; + // Fetch the latest index artifact for `subName` ("" = main index) and atomically // install it into destIndexDir. Returns true on success; on failure `err` is set // and destIndexDir is left untouched (caller may fall back to git). @@ -186,7 +195,8 @@ bool fetch_index_artifact(const std::filesystem::path& destIndexDir, std::string& err, std::string_view subName = {}, const ArtifactSource* custom = nullptr, - std::string_view pin = {}); + std::string_view pin = {}, + BytesProgress onProgress = nullptr); // Reconcile leftover index temp dirs from crashed / SIGKILL'd runs: restore an // index dir orphaned by an interrupted swap (`.old.` holding @@ -288,7 +298,8 @@ bool fetch_index_artifact(const std::filesystem::path& destIndexDir, std::string& err, std::string_view subName, const ArtifactSource* custom, - std::string_view pin); + std::string_view pin, + BytesProgress onProgress); // Reconcile leftover index temp dirs from crashed / SIGKILL'd runs. Two jobs: // 1) RECOVER: if a live index dir is missing or empty but a sibling diff --git a/src/core/xim/repo.cpp b/src/core/xim/repo.cpp index e22c83daa..a2f41233e 100644 --- a/src/core/xim/repo.cpp +++ b/src/core/xim/repo.cpp @@ -566,7 +566,8 @@ bool sync_one_repo(const IndexRepo& repo, const std::string& mirror, bool projectScope, bool force, - bool linkLocalSource) { + bool linkLocalSource, + BytesProgress onBytesProgress) { const std::string mode = repo.source.empty() ? globalIndexSource : repo.source; auto custom = artifact_source_for(repo); @@ -588,7 +589,8 @@ bool sync_one_repo(const IndexRepo& repo, std::string ferr; if (select_manifest(pointers, key, custom.has_value())) { ok = fetch_index_artifact(repoDir, ferr, key, - custom ? &*custom : nullptr, repo.version); + custom ? &*custom : nullptr, repo.version, + onBytesProgress); } else { // Say how many bases were declared only when there IS a chain. // A repo entitled through the official pointer declares none of @@ -639,7 +641,7 @@ bool sync_one_repo(const IndexRepo& repo, return true; } -bool sync_all_repos(bool force) { +bool sync_all_repos(bool force, RepoSyncProgress onProgress, BytesProgress onBytesProgress) { namespace fs = std::filesystem; auto mirror = Config::mirror(); @@ -731,21 +733,29 @@ bool sync_all_repos(bool force) { for (auto& repo : repos) if (repo.name != Config::DEFAULT_INDEX_REPO_NAME) ordered.push_back(&repo); - int total = 0, ok = 0; + // Filtered up front so the progress callback can report a true + // total -- a declared sub-index is configuration, not a step of its + // own (see above), and must not inflate the count it is reported + // against. + std::vector toSync; + toSync.reserve(ordered.size()); for (auto* repo : ordered) { - // Configuration for a declared sub-index, not a repository of its - // own: it is synced ONCE, below, with these settings applied. - // Judged after the default index has landed -- the declaration - // lives in it. if (!projectScope && classify_index_entry(*repo) == IndexEntryKind::DeclaredSubIndex) { continue; } - ++total; + toSync.push_back(repo); + } + + int ok = 0, step = 0; + const int total = static_cast(toSync.size()); + for (auto* repo : toSync) { + ++step; + if (onProgress) onProgress(repo->name, step, total); if (pastDeadline()) { ++skippedForDeadline; continue; } if (sync_one_repo(*repo, Config::repo_dir_for(*repo, projectScope), indexSource, mirror, projectScope, force, - /*linkLocalSource=*/true)) ++ok; + /*linkLocalSource=*/true, onBytesProgress)) ++ok; } return total == 0 || ok > 0; }; @@ -772,23 +782,29 @@ bool sync_all_repos(bool force) { auto subReposRoot = sub_repos_dir(); fs::create_directories(subReposRoot); - for (auto& declared : allSubRepos) { - const auto repo = with_configured_settings(declared); - if (pastDeadline()) { - ++skippedForDeadline; - // Keep it in the saved set: it was not refreshed, it was not - // withdrawn. Dropping it here would deregister a healthy - // sub-index because the clock ran out. - syncedSubRepos.push_back(declared); - continue; + { + int step = 0; + const int total = static_cast(allSubRepos.size()); + for (auto& declared : allSubRepos) { + ++step; + const auto repo = with_configured_settings(declared); + if (onProgress) onProgress(repo.name, step, total); + if (pastDeadline()) { + ++skippedForDeadline; + // Keep it in the saved set: it was not refreshed, it was not + // withdrawn. Dropping it here would deregister a healthy + // sub-index because the clock ran out. + syncedSubRepos.push_back(declared); + continue; + } + // The saved set records what is DECLARED; a pin or artifact base + // from index_repos stays in the config it came from. + if (sync_one_repo(repo, sub_repo_dir_for(repo), indexSource, mirror, false, force, + /*linkLocalSource=*/false, onBytesProgress)) + syncedSubRepos.push_back(declared); + else + log::warn("failed to sync sub-index repo: {} ({})", repo.name, repo.url); } - // The saved set records what is DECLARED; a pin or artifact base from - // index_repos stays in the config it came from. - if (sync_one_repo(repo, sub_repo_dir_for(repo), indexSource, mirror, false, force, - /*linkLocalSource=*/false)) - syncedSubRepos.push_back(declared); - else - log::warn("failed to sync sub-index repo: {} ({})", repo.name, repo.url); } save_sub_repos_json(sub_repos_json_path(), syncedSubRepos); @@ -812,7 +828,11 @@ bool sync_all_repos(bool force) { std::vector projSyncedSubs; auto projSubRoot = sub_repos_dir(true); fs::create_directories(projSubRoot); + int projStep = 0; + const int projTotal = static_cast(projMerged.size()); for (auto& [name, repo] : projMerged) { + ++projStep; + if (onProgress) onProgress(repo.name, projStep, projTotal); if (pastDeadline()) { ++skippedForDeadline; projSyncedSubs.push_back(repo); @@ -820,7 +840,7 @@ bool sync_all_repos(bool force) { } if (sync_one_repo(repo, sub_repo_dir_for(repo, true), indexSource, mirror, true, force, - /*linkLocalSource=*/false)) { + /*linkLocalSource=*/false, onBytesProgress)) { projSyncedSubs.push_back(repo); } else { log::warn("failed to sync project sub-index repo: {} ({})", repo.name, repo.url); diff --git a/src/core/xim/repo.cppm b/src/core/xim/repo.cppm index 8733a4ae8..19054d74b 100644 --- a/src/core/xim/repo.cppm +++ b/src/core/xim/repo.cppm @@ -116,7 +116,8 @@ bool sync_one_repo(const IndexRepo& repo, const std::string& mirror, bool projectScope, bool force, - bool linkLocalSource); + bool linkLocalSource, + BytesProgress onBytesProgress = nullptr); // A local repo source (file:// URL or a filesystem path) has no network host, // so the TCP reachability probe below always reports it "unreachable" and would @@ -205,9 +206,21 @@ bool sub_indexes_initialized(); // index and nowhere else. std::filesystem::path main_repo_dir(); +// Reported once per repository, right before sync_one_repo runs it: its +// name, its 1-based position, and the total for the GROUP it belongs to +// (top-level index_repos, or one batch of discovered sub-indexes). There is +// no single grand total across every group in one call -- sub-index repos +// are only discovered after the default index has synced, so a caller +// cannot know that count in advance without having done the work already. +using RepoSyncProgress = std::function; + // Sync all configured repos. // Global repos live under XLINGS_HOME/data, project repos under project .xlings/data. -bool sync_all_repos(bool force = false); +// `onBytesProgress` (see indexfetch's BytesProgress) is forwarded to whichever +// repo is being fetched as an artifact; a git clone or a local-source link has +// no byte-granular progress to report and never calls it. +bool sync_all_repos(bool force = false, RepoSyncProgress onProgress = nullptr, + BytesProgress onBytesProgress = nullptr); // Read the git HEAD hash for a repo directory. // Returns empty string for non-git repos or on any error. diff --git a/src/interface.cpp b/src/interface.cpp index e0ffe9c1f..9dcd8c7de 100644 --- a/src/interface.cpp +++ b/src/interface.cpp @@ -126,6 +126,51 @@ validate_params_(const std::string& inputSchema, const std::string& params) { "missing required field(s): " + join(missing), hint }; } +// One recognised shape for a line captured off the real stdout while a +// capability runs: "[i/n] ", the in-place progress line written by +// xim-pkgindex-*/pkgindex-build.lua while rebuilding a stale index cache +// (run in-process through libxpkg's Lua sandbox, which is why it can put raw +// text on stdout at all -- see docs/spec/interface-ndjson-v1.md §5). Not +// specific to that one script: anything else shaped like a bracketed step +// counter reads the same way, which is what makes converting it to a +// `progress` event a general answer rather than a special case for one +// vendored file. +struct BracketedStep { int index; int total; std::string message; }; + +std::optional parse_bracketed_step_(std::string_view line) { + // Strip a trailing ANSI CSI sequence (ESC '[' ). + // The one this writer emits is "\x1b[K" (clear to end of line); stripping + // any CSI is no less correct and also covers a coloured variant. + if (auto esc = line.rfind('\x1b'); esc != std::string_view::npos) { + auto rest = line.substr(esc); + std::size_t i = (rest.size() > 1 && rest[1] == '[') ? 2 : rest.size(); + while (i < rest.size() && ((rest[i] >= '0' && rest[i] <= '9') || rest[i] == ';')) ++i; + // A parameter section may be empty ("\x1b[K" has none between '[' + // and the final byte) -- what marks a valid CSI is a final byte + // present after it, not how many parameter characters were consumed. + if (i >= 2 && i < rest.size()) line = line.substr(0, esc); + } + if (line.empty() || line.front() != '[') return std::nullopt; + auto close = line.find(']'); + if (close == std::string_view::npos) return std::nullopt; + auto inner = line.substr(1, close - 1); + auto slash = inner.find('/'); + if (slash == std::string_view::npos) return std::nullopt; + + auto parse_int = [](std::string_view s, int& out) { + auto r = std::from_chars(s.data(), s.data() + s.size(), out); + return r.ec == std::errc{} && r.ptr == s.data() + s.size(); + }; + int index = 0, total = 0; + if (!parse_int(inner.substr(0, slash), index)) return std::nullopt; + if (!parse_int(inner.substr(slash + 1), total)) return std::nullopt; + if (total <= 0 || index < 0) return std::nullopt; + + auto rest = line.substr(close + 1); + while (!rest.empty() && rest.front() == ' ') rest.remove_prefix(1); + return BracketedStep{ index, total, std::string(rest) }; +} + } // namespace int run(const mcpplibs::cmdline::ParsedArgs& args, @@ -247,6 +292,31 @@ int run(const mcpplibs::cmdline::ParsedArgs& args, capability::Result result; int exit_code = 0; + // The protocol forbids anything but NDJSON on stdout while a capability + // runs. Nothing in this codebase writes raw text there any more (log:: + // and every progress renderer already gate on platform::is_tui_mode()), + // but a capability can still run in-process code this program does not + // control -- e.g. a stale index cache rebuild running libxpkg's Lua + // sandbox over a downloaded pkgindex-build.lua -- which can write + // straight to the real stdout regardless. Real fd 1 is captured for the + // whole call: a recognisable "[i/n] ..." step becomes a `progress` + // event (so a client renders it, rather than losing it); anything else + // is not left on stdout at all -- it goes to stderr instead. + platform::StdoutCapture stdoutCapture([&session](std::string_view line) { + if (auto step = parse_bracketed_step_(line)) { + session.emit_event(Event{ProgressEvent{ + .phase = "index_rebuild", + .percent = static_cast(step->index) + / static_cast(step->total), + .message = step->total > 0 + ? std::format("rebuilding index cache {}/{}: {}", + step->index, step->total, step->message) + : step->message, + }}); + } else if (!line.empty()) { + std::cerr << "[stray stdout] " << line << "\n"; + } + }); try { result = cap->execute(cap_args, stream, &token); } catch (const CancelledException&) { @@ -314,6 +384,10 @@ namespace xlings::interface { InterfaceSession::InterfaceSession(EventStream& stream, CancellationToken& token) : stream_(stream), token_(token), last_emit_(std::chrono::steady_clock::now()) { + // Taken now, before run() can install a platform::StdoutCapture around a + // capability's execution: this is the only fd every line this session + // writes will ever use, so redirecting fd 1 later cannot affect it. + wireFd_ = platform::dup_stdout_fd(); heartbeat_thread_ = std::jthread([this](std::stop_token st) { heartbeat_loop_(st); }); stdin_thread_ = std::jthread([this](std::stop_token st) { stdin_loop_(st); }); } @@ -321,6 +395,7 @@ InterfaceSession::InterfaceSession(EventStream& stream, CancellationToken& token InterfaceSession::~InterfaceSession() { if (heartbeat_thread_.joinable()) heartbeat_thread_.request_stop(); if (stdin_thread_.joinable()) stdin_thread_.request_stop(); + platform::close_fd(wireFd_); } void InterfaceSession::emit_event(const Event& e) { @@ -350,7 +425,20 @@ void InterfaceSession::emit_result(int exitCode, std::string_view raw_content) { void InterfaceSession::emit_raw_line_(std::string_view s) { std::lock_guard lock(io_mtx_); - std::cout << s << "\n" << std::flush; + // Through the duplicated fd captured at construction, not std::cout: a + // platform::StdoutCapture may have fd 1 pointed at its own pipe for the + // duration of a capability's execution, and this line must reach the + // real stdout regardless. + std::string line(s); + line += "\n"; + if (wireFd_ >= 0) { + platform::write_fd(wireFd_, line); + } else { + // dup_stdout_fd() failed at construction (fd exhaustion or an + // unsupported platform) -- fall back to std::cout rather than drop + // the line entirely. + std::cout << line << std::flush; + } last_emit_.store(std::chrono::steady_clock::now(), std::memory_order_release); } diff --git a/src/interface.cppm b/src/interface.cppm index 9277609f7..bc40b8d46 100644 --- a/src/interface.cppm +++ b/src/interface.cppm @@ -63,6 +63,12 @@ private: EventStream& stream_; CancellationToken& token_; std::mutex io_mtx_; + // A duplicate of fd 1 taken at construction time, before any capability + // runs. Every NDJSON line this session writes goes out through it + // rather than through fd 1 directly, so a platform::StdoutCapture + // installed around a capability's execution (see run()) can redirect + // fd 1 without swallowing the protocol's own output along with it. + int wireFd_ { -1 }; std::atomic saw_error_ { false }; std::atomic last_emit_; std::jthread heartbeat_thread_; diff --git a/tests/e2e/interface_update_packages_progress_test.sh b/tests/e2e/interface_update_packages_progress_test.sh new file mode 100755 index 000000000..74f0ce5af --- /dev/null +++ b/tests/e2e/interface_update_packages_progress_test.sh @@ -0,0 +1,183 @@ +#!/usr/bin/env bash +# interface_update_packages_progress_test.sh — regression test for the +# `interface update_packages` NDJSON defects found while wiring mcpp's +# index refresh through the interface protocol instead of the bare CLI: +# +# Defect 1: a refresh that rebuilds the index cache emits NOTHING but +# `heartbeat` and `result` events -- no `progress` at all, so a client +# has no way to show the user anything is happening. +# +# Defect 2 (the more serious one): when the index cache IS rebuilt, the +# rebuild runs a downloaded `pkgindex-build.lua` in-process through +# libxpkg's Lua sandbox, and that script's own progress indicator writes +# raw terminal text -- "\r[i/n] ns::name\x1b[K" -- straight onto the +# process's real stdout. The interface protocol allows nothing but one +# JSON object per line there (docs/spec/interface-ndjson-v1.md §5); a +# client reading NDJSON hits a line that is not JSON and the stream is +# no longer parseable from that point on. +# +# This fixture reproduces defect 2 without any network access: a local +# index repo whose own pkgindex-build.lua writes exactly that shape of +# text (mirroring xim-pkgindex-awesome's pkgindex-update.lua) is enough to +# force the cache rebuild, since `update_packages` always calls +# PackageCatalog::rebuild(/*force=*/true). +# +# Assertions: +# A1: every stdout line from `interface update_packages` parses as JSON. +# A2: at least one `{"kind":"progress"}` line appears in that stream. +# +# Fully self-contained/offline: builds its own local global index under the +# runtime dir; never mutates tests/fixtures. + +set -euo pipefail + +# shellcheck source=./project_test_lib.sh +source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/project_test_lib.sh" + +RUNTIME_DIR="$ROOT_DIR/tests/e2e/runtime/interface_update_packages_progress" +HOME_DIR="$RUNTIME_DIR/home" +GLOBAL_DIR="$RUNTIME_DIR/global_index" # the local global (xim) index — offline + +cleanup() { rm -rf "$RUNTIME_DIR"; } +trap cleanup EXIT +cleanup + +XLINGS_BIN="$(find_xlings_bin)" + +mkdir -p "$HOME_DIR/subos/default/bin" "$GLOBAL_DIR/pkgs/f" + +# Two package files: enough for the noisy scan's "[i/n]" counter to be +# meaningful (n > 1), not merely 1/1. +emit_pkg() { # $1=path $2=name + cat > "$1" < "$GLOBAL_DIR/xim-indexrepos.lua" + +# The fixture's own pkgindex-build.lua: runs on every FORCED rebuild +# (build_index calls run_pkgindex_build() before parsing pkgs/), through +# libxpkg's plain-Lua build sandbox (full io library, no cprintf) -- the +# same runtime xim-pkgindex-awesome's pkgindex-update.lua runs under. Its +# install() reproduces that script's progress() verbatim: a self-refreshing +# "\r[i/n] ns::name\x1b[K" line, real io.write + io.flush. +cat > "$GLOBAL_DIR/pkgindex-build.lua" <<'LUA' +package = { + name = "pkgindex-update", + namespace = "fixture", +} + +local pkgsdir = path.join(os.scriptdir(), "pkgs") + +local function progress(i, n, name) + if io and type(io.write) == "function" then + io.write(string.format("\r[%d/%d] fixture::%s", i, n, name) .. "\27[K") + if type(io.flush) == "function" then io.flush() end + else + print(string.format("[%d/%d] fixture::%s", i, n, name)) + end +end + +function installed() + return false +end + +function install() + local files = os.files(path.join(pkgsdir, "**.lua")) + local total = #files + for i, file in ipairs(files) do + progress(i, total, path.filename(file)) + end + if total > 0 then print("") end + return true +end + +function uninstall() + return true +end +LUA + +cat > "$HOME_DIR/.xlings.json" </dev/null 2>&1 || fail "self init failed" +mkdir -p "$HOME_DIR/data/xim-index-repos" +printf '{}\n' > "$HOME_DIR/data/xim-index-repos/xim-indexrepos.json" + +# ════════════════════════════════════════════════════════════════════ +# The refresh: this is what mcpp will run in place of `xlings update`. +# ════════════════════════════════════════════════════════════════════ +log "update_packages: forced rebuild runs the fixture's pkgindex-build.lua" + +STDOUT_FILE="$RUNTIME_DIR/stdout.ndjson" +STDERR_FILE="$RUNTIME_DIR/stderr.log" + +set +e +RUN_HOME interface update_packages --args '{}' \ + >"$STDOUT_FILE" 2>"$STDERR_FILE" +rc=$? +set -e + +echo "---- stdout ----"; sed 's/^/ /' "$STDOUT_FILE" +echo "---- stderr ----"; sed 's/^/ /' "$STDERR_FILE" + +[[ "$rc" -eq 0 ]] || fail "update_packages exited $rc" + +# ── A1: every stdout line parses as JSON (protocol §5: nothing else may +# reach stdout in interface mode) ────────────────────────────────── +if command -v python3 >/dev/null 2>&1; then + bad_lines=0 + while IFS= read -r line; do + [[ -n "$line" ]] || continue + if ! python3 -c 'import sys, json; json.loads(sys.argv[1])' "$line" \ + >/dev/null 2>&1; then + bad_lines=$((bad_lines + 1)) + echo "[project-e2e] not JSON: $line" >&2 + fi + done < "$STDOUT_FILE" + [[ "$bad_lines" -eq 0 ]] \ + || fail "A1: $bad_lines stdout line(s) did not parse as JSON -- raw terminal text reached stdout" + log " ✓ A1: every stdout line parses as JSON" +else + log " SKIP A1 detail check: no python3 to validate JSON (grep fallback below still runs)" + ! grep -qF $'\x1b[' "$STDOUT_FILE" \ + || fail "A1 (fallback): an ANSI escape sequence reached stdout" +fi + +# ── A2: at least one progress event was emitted for the rebuild ─────── +grep -q '"kind":"progress"' "$STDOUT_FILE" \ + || fail "A2: no progress event on the wire for a refresh that rebuilds the index" +log " ✓ A2: at least one progress event emitted" + +log "PASS: interface update_packages progress + no stray stdout text" diff --git a/tests/e2e/run_all.sh b/tests/e2e/run_all.sh index 198e51202..a62922e31 100755 --- a/tests/e2e/run_all.sh +++ b/tests/e2e/run_all.sh @@ -274,6 +274,11 @@ TESTS=( # keeps the previous one when the reinstall fails (#620); every top-level # install reports install_targets, including "already installed". "E2E-121|install_revision_test.sh||" + # `interface update_packages` reports progress for an index refresh and + # never lets a downloaded pkgindex-build.lua's raw terminal text reach + # stdout, even when the forced rebuild it triggers runs under + # install_packages instead. + "E2E-122|interface_update_packages_progress_test.sh||" ) # ── orphan check: a test that runs NOWHERE looks exactly like one that passes ── From 8b70fdf30fdc5b0d63be491535e28da6e22e9cc3 Mon Sep 17 00:00:00 2001 From: speak-agent <248744407+speak-agent@users.noreply.github.com> Date: Sun, 27 Sep 2026 22:39:13 +0800 Subject: [PATCH 2/2] chore(2026.9.28.1): bump version; interface protocol 1.2 --- docs/spec/interface-ndjson-v1.md | 7 ++++--- mcpp.toml | 2 +- src/core/config.cppm | 2 +- src/interface.cppm | 5 ++++- tests/unit/test_interface_protocol.cpp | 2 +- 5 files changed, 11 insertions(+), 7 deletions(-) diff --git a/docs/spec/interface-ndjson-v1.md b/docs/spec/interface-ndjson-v1.md index 0d2acd97e..426fa4056 100644 --- a/docs/spec/interface-ndjson-v1.md +++ b/docs/spec/interface-ndjson-v1.md @@ -1,15 +1,16 @@ > 编写日期: 2026-05-17 | 版本: 2026.9.27.1 -# NDJSON 接口协议规范 v1.1 +# NDJSON 接口协议规范 v1.2 ## 1. 概述 -`xlings interface` 提供面向程序的结构化 API，使外部客户端（IDE 插件、CI 脚本、AI agent 等）可通过标准 IO 与 xlings 交互。协议版本为 **1.1**，基于 NDJSON（Newline-Delimited JSON）。 +`xlings interface` 提供面向程序的结构化 API，使外部客户端（IDE 插件、CI 脚本、AI agent 等）可通过标准 IO 与 xlings 交互。协议版本为 **1.2**，基于 NDJSON（Newline-Delimited JSON）。 | 协议版本 | xlings | 变化 | |----------|--------|------| | 1.0 | 0.4.36 起 | 初始版本 | | 1.1 | 2026.9.27.1 起 | 增补：`install_targets` 事件；`install_plan` 条目的第三个元素 `revision`（§6.3.1） | +| 1.2 | 2026.9.28.1 起 | 增补：`update_packages` 的 `progress` 事件（`index_sync`、`index_rebuild`，§6.1.1）与 `download_progress` 数据事件（§6.3.2）；任何能力都不向 stdout 写 NDJSON 以外的内容（§5） | 次版本号的变化只做增补，1.0 客户端无需修改即可读取 1.1 的输出。客户端应通过**探测能力** 判断服务端是否提供某项功能（例如 `install_targets` 事件是否出现），而不是比较版本号。 @@ -42,7 +43,7 @@ xlings interface --version 服务端输出一行后退出： ```json -{"protocol_version":"1.1"} +{"protocol_version":"1.2"} ``` ### 3.2 查询可用能力 diff --git a/mcpp.toml b/mcpp.toml index 49fe0a76e..b9eae6912 100644 --- a/mcpp.toml +++ b/mcpp.toml @@ -29,7 +29,7 @@ members = [ [package] name = "xlings" -version = "2026.9.27.1" +version = "2026.9.28.1" description = "Universal package management infrastructure tool with SubOS isolation" license = "Apache-2.0" repo = "https://github.com/openxlings/xlings" diff --git a/src/core/config.cppm b/src/core/config.cppm index c8b459b7a..3a4fb9137 100644 --- a/src/core/config.cppm +++ b/src/core/config.cppm @@ -10,7 +10,7 @@ import xlings.core.xvm.db; namespace xlings { export struct Info { - static constexpr std::string_view VERSION = "2026.9.27.1"; + static constexpr std::string_view VERSION = "2026.9.28.1"; static constexpr std::string_view REPO = "https://github.com/openxlings/xlings"; }; diff --git a/src/interface.cppm b/src/interface.cppm index bc40b8d46..5c0798e56 100644 --- a/src/interface.cppm +++ b/src/interface.cppm @@ -26,7 +26,10 @@ namespace xlings::interface { // recipe revision as a third element of each `install_plan` entry. A 1.0 // client ignores both. Clients detect a capability by its presence on the // wire, not by comparing this string. -export constexpr const char* kProtocolVersion = "1.1"; +// 1.2 (2026.9.28.1, additive): `update_packages` reports its work as +// `progress` events (`index_sync`, `index_rebuild`) and `download_progress` +// data events, and no capability writes anything but NDJSON to stdout. +export constexpr const char* kProtocolVersion = "1.2"; // Convert any Event variant to one NDJSON line (no trailing newline). // Returns "" for events not surfaced to wire (e.g. CompletedEvent — the diff --git a/tests/unit/test_interface_protocol.cpp b/tests/unit/test_interface_protocol.cpp index 7c4455b33..5ded6e020 100644 --- a/tests/unit/test_interface_protocol.cpp +++ b/tests/unit/test_interface_protocol.cpp @@ -192,7 +192,7 @@ TEST(InterfaceProtocol, VersionFlagPrintsProtocolVersion) { auto j = nlohmann::json::parse(out, nullptr, false); ASSERT_FALSE(j.is_discarded()) << "non-JSON output: " << out; ASSERT_TRUE(j.contains("protocol_version")); - EXPECT_EQ(j["protocol_version"].get(), "1.1"); + EXPECT_EQ(j["protocol_version"].get(), "1.2"); std::filesystem::remove_all(home); }