openvibers/openvibe.tools Commits

Sandbox and developer-app results to Media under the project's namespace (roadmap WS-L task 5): a job submitted by a developer app remembers its project (tool_jobs.project_id), and its result files go to Media under tools.app.<project_id> (production) or tools.app.<project_id>.sandbox, so Media's namespace usage meters each project; sandbox results, kept only on this server before, now reach Media too. A person's or a session's job is unchanged (the tools root). A failed upload still leaves the result here and says so

OP OpenVibers committed 26 Sept 2026 · 04a5876022 · 6 files, +53 −12





Changes
ddiff --git a/apps/_shared/jobs/http.js b/apps/_shared/jobs/http.js
iindex d061e78..812cea5 100644
---- a/apps/_shared/jobs/http.js
++++ b/apps/_shared/jobs/http.js
@@@ -174,7 +174,7 @@ function mountJobRoutes(app, o) {
              if (!ok) { for (const f of files) if (f.path) fs.unlink(f.path, () => {}); return; }
          }
          try {
--            const { job, replayed } = await system.submit({ owner: w.owner, type: String(b.type || ''), input, files, idempotencyKey: key, ttlMs: ttlMs(req, w), env: w.env || 'production', ipKey: w.caller ? w.caller.ipKey : null });
++            const { job, replayed } = await system.submit({ owner: w.owner, type: String(b.type || ''), input, files, idempotencyKey: key, ttlMs: ttlMs(req, w), env: w.env || 'production', ipKey: w.caller ? w.caller.ipKey : null, project: w.kind === 'principal' && w.claims && w.claims.actor_type === 'app' ? w.claims.project_id : null });
              res.set('Location', `/api/v1/jobs/${job.id}`);
              if (replayed) res.set('Idempotent-Replayed', 'true');
              return res.status(replayed ? 200 : 202).json(system.view(job));
ddiff --git a/apps/_shared/jobs/media.js b/apps/_shared/jobs/media.js
iindex 34bf165..d1bc345 100644
---- a/apps/_shared/jobs/media.js
++++ b/apps/_shared/jobs/media.js
@@@ -64,13 +64,14 @@ function createMediaResults(o) {
          return m ? { 'X-OV-Subject': m[1] } : {};
      };
  
--    /** Store one file → MediaRef { media_id, role, namespace, size_bytes, content_hash, mime_type }. */
--    async function upload({ path: file, name, mime, size, sha256, owner, jobId, service, type }) {
++    /** Store one file → MediaRef { media_id, role, namespace, size_bytes, content_hash, mime_type }; `namespace` names a child of this one. */
++    async function upload({ path: file, name, mime, size, sha256, owner, jobId, service, type, namespace = null }) {
++        const child = namespace && String(namespace).startsWith(`${o.namespace || 'tools'}.`) ? String(namespace) : null;
          const init = await json(await call(api, {
              method: 'POST',
              headers: { 'Content-Type': 'application/json', ...subjectHeader(owner) },
              body: JSON.stringify({
--                kind: 'file', visibility: 'private', size_bytes: size, mime_type: mime, filename: name, content_hash: sha256,
++                kind: 'file', visibility: 'private', size_bytes: size, mime_type: mime, filename: name, content_hash: sha256, ...(child ? { namespace: child } : {}),
                  metadata: { source: 'openvibe.tools', service, job_id: jobId, job_type: type },
              }),
          }), 'init');
@@@ -88,7 +89,7 @@ function createMediaResults(o) {
              const done = await json(await call(`${api}/${encodeURIComponent(id)}/complete`, {
                  method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ content_hash: sha256 }),
              }), 'complete');
--            return { media_id: id, role: 'output', namespace: o.namespace || 'tools', size_bytes: size, content_hash: sha256, mime_type: mime, status: done.lifecycle_status || 'ready' };
++            return { media_id: id, role: 'output', namespace: child || o.namespace || 'tools', size_bytes: size, content_hash: sha256, mime_type: mime, status: done.lifecycle_status || 'ready' };
          } catch (err) {
              await remove(id).catch(() => {});
              throw err;
ddiff --git a/apps/_shared/jobs/store.js b/apps/_shared/jobs/store.js
iindex b1ec52f..a280e96 100644
---- a/apps/_shared/jobs/store.js
++++ b/apps/_shared/jobs/store.js
@@@ -68,6 +68,8 @@ const ADDED_COLUMNS = [
      ['ip_key', 'TEXT'],         // a browser session's job: the guard's hashed address key (never the address), so
                                  // unfinished jobs are also bounded per address (a new cookie is no new allowance)
      ['tool', 'TEXT'],           // the tool (tools.tool@1 id) whose run created the job (POST /api/v1/tools/:id/run)
++    ['project_id', 'TEXT'],     // a developer app's job: its project (prj_…), so its results go to Media under the
++                                // project's namespace, metered per project (WS-L task 5)
  ];
  
  const parse = (s, fallback) => { if (s == null) return fallback; try { return JSON.parse(s); } catch { return fallback; } };
@@@ -81,8 +83,8 @@ function createStore(db) {
      db.exec("CREATE INDEX IF NOT EXISTS tool_jobs_ip ON tool_jobs(ip_key, state) WHERE ip_key IS NOT NULL");
  
      const q = {
--        insert: db.prepare(`INSERT INTO tool_jobs (id, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool)
--            VALUES (@id, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool)`),
++        insert: db.prepare(`INSERT INTO tool_jobs (id, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id)
++            VALUES (@id, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id)`),
          markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND state = 'failed' AND retried_by IS NULL"),
          get: db.prepare('SELECT * FROM tool_jobs WHERE id = ?'),
          byIdem: db.prepare('SELECT * FROM tool_jobs WHERE owner = ? AND idempotency_key = ?'),
@@@ -127,7 +129,7 @@ function createStore(db) {
      return {
          db,
          TERMINAL, STATES,
--        insert(row) { q.insert.run({ retry_of: null, ip_key: null, tool: null, ...row }); },
++        insert(row) { q.insert.run({ retry_of: null, ip_key: null, tool: null, project_id: null, ...row }); },
          byResultMedia: (owner, mediaId) => q.byResultMedia.all(owner, `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%`),
          markRetried: (id, next, now = Date.now()) => q.markRetried.run({ id, next, now }).changes === 1,
          get: (id) => q.get.get(id) || null,
ddiff --git a/apps/_shared/jobs/system.js b/apps/_shared/jobs/system.js
iindex af1eeb1..3352571 100644
---- a/apps/_shared/jobs/system.js
++++ b/apps/_shared/jobs/system.js
@@@ -38,6 +38,15 @@ const { createStore, TERMINAL } = require('./store');
  const { createJobEvents } = require('./events');
  
  const HOUR = 60 * 60 * 1000;
++
++// A developer project's jobs (WS-L task 5): results go to Media under <tools namespace>.app.<project_id>
++// (production) or .app.<project_id>.sandbox, so Media's namespace usage meters each project.
++const PROJECT_RE = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;
++const projectOf = (p) => (PROJECT_RE.test(String(p || '')) ? String(p) : null);
++function projectNamespace(media, row) {
++    if (!row.project_id || !PROJECT_RE.test(row.project_id)) return null;
++    return `${media.namespace || 'tools'}.app.${row.project_id}${row.env === 'sandbox' ? '.sandbox' : ''}`;
++}
  const PRUNE_RECHECK_MS = 24 * HOUR;   // an expired job the pruner could not finish is looked at again after this
  const REF_RE = /^[a-z][a-z0-9_-]*(?::[A-Za-z0-9_.-]+){1,4}$/;   // <service>:<kind>:<id>, e.g. community:paste:p_123
  const MAX_REFS = 50;
@@@ -203,7 +212,7 @@ function createJobSystem(o) {
      // Developer-app sandbox jobs: fewer at a time, kept briefly, results never leave this server.
      const SANDBOX_MAX_ACTIVE = Math.max(1, o.sandboxMaxActivePerOwner || 2);
      const SANDBOX_TTL_MS = 30 * 60 * 1000;
--    async function submit({ owner, type, input = {}, files = [], idempotencyKey = null, ttlMs = HOUR, env = 'production', ipKey = null, tool = null }) {
++    async function submit({ owner, type, input = {}, files = [], idempotencyKey = null, ttlMs = HOUR, env = 'production', ipKey = null, tool = null, project = null }) {
          if (env !== 'sandbox') env = 'production';
          if (env === 'sandbox') ttlMs = Math.min(ttlMs, SANDBOX_TTL_MS);
          if (stopped) throw new JobError(503, 'tools.job.unavailable', 'The job system is shutting down');
@@@ -254,6 +263,7 @@ function createJobSystem(o) {
                          max_attempts: def.maxAttempts, ttl_ms: ttlMs, now: Date.now(), env,
                          ip_key: String(owner).startsWith('session:') && ipKey ? String(ipKey) : null,
                          tool: tool && /^[a-z][a-z0-9-]{0,39}$/.test(String(tool)) ? String(tool) : null,
++                        project_id: projectOf(project),
                      });
                      announce.created(store.get(id));
                  })();
@@@ -389,11 +399,13 @@ function createJobSystem(o) {
              const size = fs.statSync(local).size;
              const sha256 = await sha256File(local);
              const rec = { name: String(f.name || `result-${i}`).slice(0, 200), mime: f.mime || 'application/octet-stream', size, sha256, storage: 'local', media: null, _path: local };
--            if (media && !entry.reason && row.env !== 'sandbox') {
++            // A developer app's results (sandbox too) go under its project's namespace (WS-L task 5); other
++            // sandbox results stay on this server until the job expires.
++            if (media && !entry.reason && (row.env !== 'sandbox' || row.project_id)) {
                  try {
                      const progress = progressFn(row.id, entry);
                      progress(99, 'Storing the result');
--                    rec.media = await media.upload({ path: local, name: rec.name, mime: rec.mime, size, sha256, owner: row.owner, jobId: row.id, service: o.service, type: row.type });
++                    rec.media = await media.upload({ path: local, name: rec.name, mime: rec.mime, size, sha256, owner: row.owner, jobId: row.id, service: o.service, type: row.type, namespace: projectNamespace(media, row) });
                      rec.storage = 'media';
                      delete rec._path;
                      await fsp.unlink(local).catch(() => {});
@@@ -476,7 +488,7 @@ function createJobSystem(o) {
                  store.insert({
                      id: next, type: row.type, type_version: def.version, owner: row.owner, input_json: row.input_json, files_json: JSON.stringify(stored),
                      idempotency_key: null, request_hash: row.request_hash, max_attempts: def.maxAttempts, ttl_ms: row.ttl_ms, now: Date.now(),
--                    env: row.env, retry_of: row.id, ip_key: row.ip_key || null, tool: row.tool || null,
++                    env: row.env, retry_of: row.id, ip_key: row.ip_key || null, tool: row.tool || null, project_id: row.project_id || null,
                  });
                  announce.created(store.get(next));
              })();
ddiff --git a/apps/_shared/test/jobs.test.js b/apps/_shared/test/jobs.test.js
iindex 122b7d0..29c8536 100644
---- a/apps/_shared/test/jobs.test.js
++++ b/apps/_shared/test/jobs.test.js
@@@ -515,6 +515,31 @@ async function readSse(url, { headers = {}, stop = () => false, ms = 5000 } = {}
          assert.strictEqual(r.status, 401, 'a sandbox token that is not an app is refused'); assert.strictEqual(r.body.code, 'token.sandbox_refused');
          await s.close();
  
++        // ── A developer app's results go to Media under its project's namespace (WS-L task 5) ───
++        {
++            const fm2 = fakeMedia();
++            const ms2 = await new Promise(res => { const srv = fm2.app.listen(0, '127.0.0.1', () => res(srv)); });
++            s = await satellite(path.join(root, 'g'), { media: jobs.createMediaResults({ internalUrl: `http://127.0.0.1:${ms2.address().port}`, tokens: fakeTokens }) });
++            const PRJ = 'prj_01JCCCCCCCCCCCCCCCCCCCCCCC';
++            const done = async (c, id) => until(async () => { const g = await c(`/api/v1/jobs/${id}`); return g.body.state === 'succeeded' && g.body; }, 'project job');
++            const prod = client(s.base, { bearer: token({}) });
++            let j = await done(prod, (await prod('/api/v1/jobs', json({ type: 'test.upper', input: { text: 'prod' } }))).body.id);
++            assert.strictEqual(s.system.get(j.id).project_id, PRJ, 'the job remembers its project');
++            assert.strictEqual(j.result.files[0].storage, 'media');
++            assert.strictEqual(fm2.objects.get(j.result.files[0].media.media_id).meta.namespace, `tools.app.${PRJ}`, 'production: tools.app.<project>');
++            assert.strictEqual(j.result.files[0].media.namespace, `tools.app.${PRJ}`);
++            const sand = client(s.base, { bearer: token({ env: 'sandbox', cap: ['tools.job.create', 'tools.job.read'] }) });
++            j = await done(sand, (await sand('/api/v1/jobs', json({ type: 'test.upper', input: { text: 'sandbox' } }))).body.id);
++            assert.strictEqual(j.result.files[0].storage, 'media', 'sandbox results reach Media too, metered per project');
++            assert.strictEqual(fm2.objects.get(j.result.files[0].media.media_id).meta.namespace, `tools.app.${PRJ}.sandbox`);
++            const person = client(s.base, { user: USER_B });
++            j = await done(person, (await person('/api/v1/jobs', json({ type: 'test.upper', input: { text: 'person' } }))).body.id);
++            assert.strictEqual(fm2.objects.get(j.result.files[0].media.media_id).meta.namespace, undefined, "a person's job stays in the tools root");
++            assert.strictEqual(s.system.get(j.id).project_id, null);
++            await s.close();
++            ms2.close();
++        }
++
          console.log('jobs (shared runtime): all checks passed');
      } finally {
          fs.rmSync(root, { recursive: true, force: true });
ddiff --git a/apps/_shared/tools/run.js b/apps/_shared/tools/run.js
iindex 08b4bd3..ecda228 100644
---- a/apps/_shared/tools/run.js
++++ b/apps/_shared/tools/run.js
@@@ -410,6 +410,7 @@ function createRunApi(o) {
                      files: files.map(f => ({ path: f.path, buffer: f.buffer, name: f.originalname || f.name, mime: f.mimetype || f.mime, size: f.size })),
                      idempotencyKey: key, ttlMs: who.kind === 'session' ? 60 * 60 * 1000 : 24 * 60 * 60 * 1000,
                      env: who.env || 'production', ipKey: who.ipKey, tool: d.id,
++                    project: who.kind === 'principal' && who.claims && who.claims.actor_type === 'app' ? who.claims.project_id : null,
                  });
                  row = out.job; replayed = out.replayed;
              } catch (err) {