diff --git a/docs/spec/interface-ndjson-v1.md b/docs/spec/interface-ndjson-v1.md index 8fde847a2..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 查询可用能力 @@ -122,7 +123,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 +147,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 +252,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/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/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/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/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..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 @@ -63,6 +66,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 ── 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); }