chore: v2.8.13 — Artifacts reinstall/edit/installs, agent steps in chat, pick a local file to attach, Windows paths as links
- Chat: an Agent step shows its task, status and tool calls and opens its conversation in place; Windows paths (E:\…, C:/…) in replies open as files - Prompt: attach a file on this computer by path (native picker), clearer too-large messages - Artifacts: edit title/description and a skill's files (new snapshot), Reinstall, install history, Hub-only rows, skills installed into ~/.claude/skills and ~/.agents/skills - Files: batch upload of small files in one extract - AI Tools: npm install falls back to --prefix ~/.local when the global prefix is not writable
This commit is contained in:
+351
-2
@@ -4,7 +4,8 @@
|
||||
* or into another of the same account), delete into a per-account TRASH that
|
||||
* can be restored (kept TRASH_DAYS) — and, for `/api/homes`, one recycle bin
|
||||
* per USER over every store their deletes went to — upload a file streamed from the request
|
||||
* into any (new) sub-folder, and download files or folders as one streamed zip.
|
||||
* into any (new) sub-folder (or many small ones as ONE batch request), and
|
||||
* download files or folders as one streamed zip.
|
||||
*
|
||||
* Every path is the read path's own (git-diff.js): `lanePath` (lexical, no
|
||||
* `..`), the PARENT resolved by `resolveInLane` (realpath-confined, so no
|
||||
@@ -37,7 +38,8 @@ const crypto = require("node:crypto");
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const zlib = require("node:zlib");
|
||||
const { Readable, Transform } = require("node:stream");
|
||||
const { PassThrough, Readable, Transform } = require("node:stream");
|
||||
const { once } = require("node:events");
|
||||
const { AsyncLocalStorage } = require("node:async_hooks");
|
||||
const { pipeline } = require("node:stream/promises");
|
||||
|
||||
@@ -706,6 +708,351 @@ async function upload(dir, rel, stream, { conflict, length, trash } = {}) {
|
||||
return { path: loc.rel };
|
||||
}
|
||||
|
||||
// ── Batch upload ───────────────────────────────────────────────────────────
|
||||
//
|
||||
// Many small files cost per FILE, not per byte: one request, one path walk
|
||||
// and — in a member's folder — one sudo write-as each. A batch carries them
|
||||
// in ONE request: a 4-byte big-endian length, that many bytes of JSON
|
||||
// `{files: [{path, size, mtime?}]}` (paths relative to `?dir=`), then each
|
||||
// file's bytes in that order. Every name is checked like `upload`'s and the
|
||||
// whole batch is planned (folders, conflicts) before the first write; a
|
||||
// member's batch is written by ONE `write-as <dir> extract` from a tar this
|
||||
// module writes itself (only folders and regular files, names it checked).
|
||||
|
||||
const BATCH_MAX_FILES = 5000;
|
||||
/** A file in a batch is held in memory whole; bigger ones go through `upload`. */
|
||||
const BATCH_FILE_MAX = 16 * 1024 * 1024;
|
||||
const BATCH_MAX_BYTES = 256 * 1024 * 1024;
|
||||
const BATCH_HEAD_MAX = 4 * 1024 * 1024;
|
||||
|
||||
/** Whether a member's batch can be written (write-as `extract`, install-system re-run). */
|
||||
function needExtract(acct) {
|
||||
if (acct && !runAs.writeAsCanExtract()) {
|
||||
throw bad(
|
||||
"EUPGRADE",
|
||||
"uploading many files at once into a member's folder needs the updated helper: run `agenthub server install-system` once"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** Reads exact byte counts from a stream (the request body). */
|
||||
function byteReader(stream) {
|
||||
const it = stream[Symbol.asyncIterator]();
|
||||
let chunks = [];
|
||||
let have = 0;
|
||||
return {
|
||||
async read(n) {
|
||||
while (have < n) {
|
||||
const { value, done } = await it.next();
|
||||
if (done) throw bad("EBADBODY", "the upload ended early");
|
||||
const b = Buffer.isBuffer(value) ? value : Buffer.from(value);
|
||||
chunks.push(b);
|
||||
have += b.length;
|
||||
}
|
||||
const all = chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, have);
|
||||
const rest = all.subarray(n);
|
||||
chunks = rest.length ? [rest] : [];
|
||||
have = rest.length;
|
||||
return all.subarray(0, n);
|
||||
},
|
||||
/** Consume whatever is left, so an answer can still be sent. */
|
||||
async drain() {
|
||||
chunks = [];
|
||||
have = 0;
|
||||
for (;;) if ((await it.next()).done) return;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** The `files` list of a batch, each `{path, size, mtime}` checked for shape. */
|
||||
function batchFiles(list) {
|
||||
if (!Array.isArray(list) || !list.length) throw bad("EBADBODY", "a batch names its files");
|
||||
if (list.length > BATCH_MAX_FILES) {
|
||||
throw bad("ETOOBIG", `a batch holds at most ${BATCH_MAX_FILES} files`);
|
||||
}
|
||||
let total = 0;
|
||||
return list.map((f) => {
|
||||
const size = f?.size ?? 0;
|
||||
if (typeof f?.path !== "string" || !Number.isSafeInteger(size) || size < 0) {
|
||||
throw bad("EBADBODY", "each file is {path, size}");
|
||||
}
|
||||
if (size > BATCH_FILE_MAX) {
|
||||
throw bad("ETOOBIG", `a file in a batch may be at most ${BATCH_FILE_MAX / 1024 / 1024} MB`);
|
||||
}
|
||||
total += size;
|
||||
if (total > BATCH_MAX_BYTES) {
|
||||
throw bad("ETOOBIG", `a batch may be at most ${BATCH_MAX_BYTES / 1024 / 1024} MB`);
|
||||
}
|
||||
const mtime = Number.isFinite(f.mtime) && f.mtime > 0 ? f.mtime : Date.now();
|
||||
return { path: f.path, size, mtime };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Where each file of a batch goes, under `conflict`: the folders to make
|
||||
* (parents first) and per file its target, or `skip`/`replace`. `taken`
|
||||
* lists the names already there (paths relative to the batch's folder);
|
||||
* under `fail` they are not resolved — the caller refuses or reports them.
|
||||
*/
|
||||
async function planBatch(dir, dirRel, files, conflict) {
|
||||
const base = await folder(dir, dirRel || "");
|
||||
if (gitInside(base.rel)) throw bad("EBADPATH", "nothing is uploaded into .git");
|
||||
const full = (r) => (base.rel ? `${base.rel}/${r}` : r);
|
||||
// rel under base → {rel (workspace), abs, missing}
|
||||
const dirs = new Map([["", { rel: base.rel, abs: base.abs, missing: false }]]);
|
||||
const make = [];
|
||||
const dirOf = async (r) => {
|
||||
if (dirs.has(r)) return dirs.get(r);
|
||||
const upRel = path.posix.dirname(r) === "." ? "" : path.posix.dirname(r);
|
||||
const up = await dirOf(upRel);
|
||||
const name = path.posix.basename(r);
|
||||
const d = { rel: full(r), abs: path.join(up.abs, name), missing: true };
|
||||
if (up.missing) {
|
||||
if (topDot(base.root, d.abs)) throw hidden();
|
||||
if (await isSecretOnDisk(dir, d.rel)) throw bad("ESECRET", "that path holds secrets");
|
||||
} else {
|
||||
const loc = await locate(dir, d.rel);
|
||||
if (loc.st && !loc.st.isDirectory()) {
|
||||
throw bad("EEXISTS", `${d.rel} is a file, not a folder`, { names: [r] });
|
||||
}
|
||||
d.missing = !loc.st;
|
||||
}
|
||||
if (d.missing) make.push(d);
|
||||
dirs.set(r, d);
|
||||
return d;
|
||||
};
|
||||
|
||||
const seen = new Set();
|
||||
// A free name for `keep` is free on disk AND not another file of this batch.
|
||||
const claimed = new Set();
|
||||
const own = new Set(
|
||||
files.map((f) => {
|
||||
try {
|
||||
return lanePath(f.path);
|
||||
} catch {
|
||||
return "";
|
||||
}
|
||||
})
|
||||
);
|
||||
const out = [];
|
||||
const taken = [];
|
||||
for (const f of files) {
|
||||
const r = lanePath(f.path);
|
||||
for (const seg of r.split("/")) validName(seg);
|
||||
if (gitInside(full(r))) throw bad("EBADPATH", "nothing is uploaded into .git");
|
||||
if (seen.has(r)) throw bad("EBADPATH", `${r} is named twice`);
|
||||
seen.add(r);
|
||||
const upRel = path.posix.dirname(r) === "." ? "" : path.posix.dirname(r);
|
||||
const up = await dirOf(upRel);
|
||||
let name = path.posix.basename(r);
|
||||
let st = null;
|
||||
if (up.missing) {
|
||||
if (await isSecretOnDisk(dir, full(r))) throw bad("ESECRET", "that path holds secrets");
|
||||
} else {
|
||||
const loc = await locate(dir, full(r));
|
||||
st = loc.st;
|
||||
}
|
||||
const step = { ...f, rel: r, name, up, skip: false, replace: false };
|
||||
if (st) {
|
||||
if (st.isDirectory()) throw bad("EEXISTS", `${full(r)} is a folder`, { names: [r] });
|
||||
taken.push(r);
|
||||
if (conflict === "skip") step.skip = true;
|
||||
else if (conflict === "replace") step.replace = true;
|
||||
else if (conflict === "keep") {
|
||||
const ext = name.startsWith(".") ? "" : path.extname(name);
|
||||
const stem = ext ? name.slice(0, -ext.length) : name;
|
||||
for (let i = 1; ; i++) {
|
||||
if (i >= 10_000) throw bad("EEXISTS", `no free name for ${name}`);
|
||||
const n = `${stem} (${i})${ext}`;
|
||||
const abs = path.join(up.abs, n);
|
||||
const nRel = upRel ? `${upRel}/${n}` : n;
|
||||
if (claimed.has(abs) || own.has(nRel)) continue;
|
||||
if (await fs.promises.lstat(abs).catch(() => null)) continue;
|
||||
name = n;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
step.name = name;
|
||||
step.abs = path.join(up.abs, name);
|
||||
step.full = up.rel ? `${up.rel}/${name}` : name;
|
||||
claimed.add(step.abs);
|
||||
out.push(step);
|
||||
}
|
||||
for (const r of seen) {
|
||||
if (dirs.has(r)) throw bad("EBADPATH", `${r} is both a file and a folder`);
|
||||
}
|
||||
return { base, make, files: out, taken };
|
||||
}
|
||||
|
||||
/**
|
||||
* `{taken, batch}` for a batch about to be sent: the names already in
|
||||
* `dirRel` (relative to it), and whether a batch can be written there at all
|
||||
* (false: send the files one by one).
|
||||
*/
|
||||
async function uploadCheck(dir, dirRel, paths) {
|
||||
if (!Array.isArray(paths) || !paths.every((p) => typeof p === "string")) {
|
||||
throw bad("EBADBODY", "paths is a list of relative paths");
|
||||
}
|
||||
const files = batchFiles(paths.map((p) => ({ path: p, size: 0 })));
|
||||
const plan = await planBatch(dir, dirRel, files, "fail");
|
||||
let batch = true;
|
||||
try {
|
||||
needExtract(accountAt(plan.base.abs));
|
||||
} catch {
|
||||
batch = false;
|
||||
}
|
||||
return { taken: plan.taken, batch };
|
||||
}
|
||||
|
||||
/** One octal tar field of `len` bytes. */
|
||||
function tarOctal(buf, off, len, n) {
|
||||
buf.write(`${n.toString(8).padStart(len - 1, "0")}\0`, off, len, "ascii");
|
||||
}
|
||||
|
||||
/** A ustar header block for `name` (a pax `path` record first when it is long). */
|
||||
function tarHeader(name, { type, size, mode, mtime }) {
|
||||
const block = (n, t, sz) => {
|
||||
const h = Buffer.alloc(512);
|
||||
h.write(n, 0, 100, "utf8");
|
||||
tarOctal(h, 100, 8, mode);
|
||||
tarOctal(h, 108, 8, 0);
|
||||
tarOctal(h, 116, 8, 0);
|
||||
tarOctal(h, 124, 12, sz);
|
||||
tarOctal(h, 136, 12, Math.floor(mtime / 1000));
|
||||
h.fill(" ", 148, 156);
|
||||
h.write(t, 156, 1, "ascii");
|
||||
h.write("ustar\0" + "00", 257, 8, "ascii");
|
||||
let sum = 0;
|
||||
for (const b of h) sum += b;
|
||||
h.write(`${sum.toString(8).padStart(6, "0")}\0 `, 148, 8, "ascii");
|
||||
return h;
|
||||
};
|
||||
if (Buffer.byteLength(name) <= 100) return block(name, type, size);
|
||||
// "<len> path=<name>\n", where <len> counts its own digits.
|
||||
const body = ` path=${name}\n`;
|
||||
let len = Buffer.byteLength(body);
|
||||
while (String(len).length + Buffer.byteLength(body) !== len) {
|
||||
len = String(len).length + Buffer.byteLength(body);
|
||||
}
|
||||
const rec = Buffer.from(`${len}${body}`, "utf8");
|
||||
return Buffer.concat([block("PaxHeader", "x", rec.length), tarPad(rec), block("", type, size)]);
|
||||
}
|
||||
|
||||
function tarPad(buf) {
|
||||
const rest = buf.length % 512;
|
||||
return rest ? Buffer.concat([buf, Buffer.alloc(512 - rest)]) : buf;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write a batch streamed from `stream` (see above) into `dirRel`. Returns
|
||||
* `{saved, skipped, paths}`; under `conflict: "fail"` a taken name is 409
|
||||
* EEXISTS with every taken path in `names`.
|
||||
*/
|
||||
async function uploadBatch(dir, dirRel, stream, { conflict, trash } = {}) {
|
||||
const reader = byteReader(stream);
|
||||
try {
|
||||
const n = (await reader.read(4)).readUInt32BE(0);
|
||||
if (n > BATCH_HEAD_MAX) throw bad("ETOOBIG", "a batch's file list is too long");
|
||||
let head;
|
||||
try {
|
||||
head = JSON.parse((await reader.read(n)).toString("utf8"));
|
||||
} catch (e) {
|
||||
if (e.code) throw e;
|
||||
throw bad("EBADBODY", "a batch starts with its file list as JSON");
|
||||
}
|
||||
const files = batchFiles(head?.files);
|
||||
const policy = conflictOf(conflict);
|
||||
const plan = await planBatch(dir, dirRel, files, policy);
|
||||
if (policy === "fail" && plan.taken.length) {
|
||||
throw bad("EEXISTS", `${plan.taken.length} name(s) already exist`, { names: plan.taken });
|
||||
}
|
||||
const acct = accountAt(plan.base.abs);
|
||||
needExtract(acct);
|
||||
for (const f of plan.files) {
|
||||
if (f.replace) await trash(f.full);
|
||||
}
|
||||
const done = [];
|
||||
let skipped = 0;
|
||||
if (acct) {
|
||||
const tar = new PassThrough();
|
||||
const put = async (buf) => {
|
||||
// write-as gone (it failed): a destroyed stream never drains.
|
||||
if (tar.destroyed) throw bad("EWRITEAS", "the batch could not be written");
|
||||
if (!tar.write(buf)) await once(tar, "drain");
|
||||
};
|
||||
const now = Date.now();
|
||||
const extract = runAs.writeAsAsync(acct, plan.base.abs, ["extract"], { input: tar });
|
||||
const feed = (async () => {
|
||||
for (const d of plan.make) {
|
||||
const name = `${path.relative(plan.base.abs, d.abs).split(path.sep).join("/")}/`;
|
||||
await put(tarHeader(name, { type: "5", size: 0, mode: 0o755, mtime: now }));
|
||||
}
|
||||
for (const f of plan.files) {
|
||||
const data = await reader.read(f.size);
|
||||
if (f.skip) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
const name = path.relative(plan.base.abs, f.abs).split(path.sep).join("/");
|
||||
await put(tarHeader(name, { type: "0", size: f.size, mode: 0o644, mtime: f.mtime }));
|
||||
await put(tarPad(data));
|
||||
done.push(f.full);
|
||||
}
|
||||
tar.end(Buffer.alloc(1024));
|
||||
})();
|
||||
await Promise.all([
|
||||
feed.catch((e) => {
|
||||
tar.destroy(e);
|
||||
throw e;
|
||||
}),
|
||||
extract.catch((e) => {
|
||||
tar.destroy(e);
|
||||
throw e;
|
||||
}),
|
||||
]);
|
||||
} else {
|
||||
for (const d of plan.make) {
|
||||
await fs.promises.mkdir(d.abs).catch(async (e) => {
|
||||
const st = e.code === "EEXIST" && (await fs.promises.lstat(d.abs).catch(() => null));
|
||||
if (!st?.isDirectory()) throw e;
|
||||
});
|
||||
}
|
||||
for (const f of plan.files) {
|
||||
const data = await reader.read(f.size);
|
||||
if (f.skip) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
const tmp = path.join(
|
||||
f.up.abs,
|
||||
`.${f.name}.agenthub-upload-${crypto.randomBytes(4).toString("hex")}`
|
||||
);
|
||||
try {
|
||||
await fs.promises.writeFile(tmp, data, { flag: "wx" });
|
||||
const t = new Date(f.mtime);
|
||||
await fs.promises.utimes(tmp, t, t);
|
||||
// Linked, not renamed: a name taken meanwhile is refused, never replaced.
|
||||
await fs.promises.link(tmp, f.abs).catch(async (e) => {
|
||||
if (e.code === "EEXIST") {
|
||||
throw bad("EEXISTS", `${f.rel} already exists`, { names: [f.rel] });
|
||||
}
|
||||
if (e.code === "EPERM" || e.code === "ENOTSUP") return fs.promises.rename(tmp, f.abs);
|
||||
throw e;
|
||||
});
|
||||
} finally {
|
||||
await fs.promises.rm(tmp, { force: true });
|
||||
}
|
||||
done.push(f.full);
|
||||
}
|
||||
}
|
||||
return { saved: done.length, skipped, paths: done };
|
||||
} catch (e) {
|
||||
await reader.drain().catch(() => {});
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* One file for a download (the Files tab's `openDownload`, after this
|
||||
* module's own checks — the home rule included, on the file's REAL path).
|
||||
@@ -893,6 +1240,8 @@ module.exports = {
|
||||
restore,
|
||||
purgeTrash,
|
||||
upload,
|
||||
uploadBatch,
|
||||
uploadCheck,
|
||||
zip,
|
||||
collect,
|
||||
UPLOAD_MAX_BYTES,
|
||||
|
||||
Reference in New Issue
Block a user