openvibers/openvibe.ai Commits
Validate placement offers and requirements before planning (#9)
Changes
@@@ -21,6 +21,7 @@* after that one half-open probe decides whether it closes again.*/const placement = require('openvibe-sdk/placement');++const contracts = require('openvibe-contracts');const { AiError, resolveSecret, sha256 } = require('../util');const { retryable, sleep, ProviderError } = require('./common');const { createStubProvider } = require('./stub');@@@ -178,8 +179,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nooffers.push({offer_id: id, kind: 'provider', provider: p.key, region: 'global',trust: (p.metadata && p.metadata.local) || p.kind === 'stub' ? 'community' : 'first-party',-- capabilities: caps, latency_ms: st.p95 ? { p95: st.p95 } : {}, health: { status },-- pricing: cards ? { model: 'metered', rate_card: cards.in.id } : { model: 'prepaid' },++ capabilities: caps.map(f => typeof f === 'string' ? `ai:${f}` : f),++ latency_ms: st.p95 ? { p95: st.p95 } : {}, health: { status },++ pricing: cards ? { model: 'per-operation', rate_card: cards.in.id } : { model: 'prepaid' },++ updated_at: p.updated_at,});if (cards) {rateCards.push(cards.in, cards.cached, cards.out);@@@ -202,12 +205,20 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noconst c = route.constraints || {};const requirements = {kind: 'ai', mobility: 'request', latency_class: c.latency_class || 'interactive', objective: c.objective || 'balanced',-- capabilities: features, units: Number(c.units) || 1, latency_op: operation,++ capabilities: features.map(f => typeof f === 'string' ? `ai:${f}` : f), units: Number(c.units) || 1, latency_op: operation,...(c.max_latency_ms != null ? { max_latency_ms: Number(c.max_latency_ms) } : {}),...(c.max_cost_usd != null ? { max_cost_usd: Number(c.max_cost_usd) } : {}),...(c.trust ? { trust: c.trust } : {}),...(c.authority ? { authority: c.authority } : {}),};++ for (const offer of offers) {++ const validation = contracts.validate('platform.resource-offer@1', offer);++ if (!validation.valid) throw new Error(`invalid platform.resource-offer@1 ${offer.offer_id}: ${JSON.stringify(validation.errors)}`);++ }++ // These are SDK planner options, not fields in the workload contract.++ const { units, latency_op, authority, ...contractRequirements } = requirements;++ const validation = contracts.validate('platform.workload-requirements@1', contractRequirements);++ if (!validation.valid) throw new Error(`invalid platform.workload-requirements@1: ${JSON.stringify(validation.errors)}`);const state = await getRouteState.get(route.key);const current = state && state.current_provider ? offerId(state.current_provider, state.current_model) : null;const states = await providerStates(cardsByProvider);@@@ -96,7 +96,21 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open oassert.ok(byProvider.cheap.cost < byProvider.pricey.cost, 'rate cards carry the real prices');// explain is AI's projection of the placement result; that result is a platform.placement-result@1.-- const placed = await h.pool.candidates(r, ['classify'], 'classify');++ const validated = [];++ const originalValidate = contracts.validate;++ let placed;++ contracts.validate = (name, value) => {++ const result = originalValidate(name, value);++ validated.push({ name, value, result });++ return result;++ };++ try { placed = await h.pool.candidates(r, ['classify'], 'classify'); }++ finally { contracts.validate = originalValidate; }++ const checkedOffers = validated.filter((x) => x.name === 'platform.resource-offer@1');++ const checkedRequirements = validated.filter((x) => x.name === 'platform.workload-requirements@1');++ assert.deepStrictEqual(checkedOffers.map((x) => x.value.offer_id).sort(), placed.placement.candidates.map((x) => x.id).sort(), 'every offer was validated before planning');++ assert.strictEqual(checkedRequirements.length, 1, 'requirements were validated before planning');++ for (const x of [...checkedOffers, ...checkedRequirements]) assert.ok(x.result.valid, `${x.name}: ${JSON.stringify(x.result.errors)}`);const v = contracts.validate('platform.placement-result@1', placed.placement);assert.ok(v.valid, `platform.placement-result@1: ${JSON.stringify(v.errors)}`);assert.strictEqual(placed.placement.selected, 'cheap:cheap-m');@@@ -124,6 +138,38 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open oawait ok.close(); await flaky.close();});++t.test('an invalid offer fails closed before reaching the planner', async () => {++ const r = await h.registry.resolveRoute('test.pool-cheap');++ const originalListProviders = h.registry.listProviders;++ const placement = require('openvibe-sdk/placement');++ const originalPlan = placement.plan;++ let plannerCalls = 0;++ h.registry.listProviders = async (...args) => (await originalListProviders.apply(h.registry, args))++ .map((p) => p.key === 'cheap' ? { ...p, capabilities: ['classify', 42] } : p);++ placement.plan = () => { plannerCalls++; throw new Error('invalid offer reached planner'); };++ try {++ await assert.rejects(h.pool.candidates(r, ['classify'], 'classify'), /invalid platform\.resource-offer@1/);++ assert.strictEqual(plannerCalls, 0, 'planner was not called with the invalid offer');++ } finally {++ h.registry.listProviders = originalListProviders;++ placement.plan = originalPlan;++ }++});++++t.test('invalid requirements fail closed before reaching the planner', async () => {++ const r = await h.registry.resolveRoute('test.pool-cheap');++ const placement = require('openvibe-sdk/placement');++ const originalPlan = placement.plan;++ let plannerCalls = 0;++ placement.plan = () => { plannerCalls++; throw new Error('invalid requirements reached planner'); };++ try {++ await assert.rejects(h.pool.candidates(r, ['classify', 42], 'classify'), /invalid platform\.workload-requirements@1/);++ assert.strictEqual(plannerCalls, 0, 'planner was not called with invalid requirements');++ } finally {++ placement.plan = originalPlan;++ }++});++t.test('an upstream 429 shifts to another provider without opening the circuit', async () => {const ok = await seamServer(() => ({ text: 'steady answer', model: 'steady-m', usage: { input: 3, output: 1 } }));const busy = await seamServer(() => ({ status: 429, body: { error: { message: 'rate limited' } } }));
@@ -21,6 +21,7 @@
@@ -21,6 +21,7 @@
* after that one half-open probe decides whether it closes again.
* after that one half-open probe decides whether it closes again.
*/
*/
const placement = require('openvibe-sdk/placement');
const placement = require('openvibe-sdk/placement');
+const contracts = require('openvibe-contracts');
const { AiError, resolveSecret, sha256 } = require('../util');
const { AiError, resolveSecret, sha256 } = require('../util');
const { retryable, sleep, ProviderError } = require('./common');
const { retryable, sleep, ProviderError } = require('./common');
const { createStubProvider } = require('./stub');
const { createStubProvider } = require('./stub');
@@ -178,8 +179,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -178,8 +179,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
offers.push({
offers.push({
offer_id: id, kind: 'provider', provider: p.key, region: 'global',
offer_id: id, kind: 'provider', provider: p.key, region: 'global',
trust: (p.metadata && p.metadata.local) || p.kind === 'stub' ? 'community' : 'first-party',
trust: (p.metadata && p.metadata.local) || p.kind === 'stub' ? 'community' : 'first-party',
- capabilities: caps, latency_ms: st.p95 ? { p95: st.p95 } : {}, health: { status },
- pricing: cards ? { model: 'metered', rate_card: cards.in.id } : { model: 'prepaid' },
+ capabilities: caps.map(f => typeof f === 'string' ? `ai:${f}` : f),
+ latency_ms: st.p95 ? { p95: st.p95 } : {}, health: { status },
+ pricing: cards ? { model: 'per-operation', rate_card: cards.in.id } : { model: 'prepaid' },
+ updated_at: p.updated_at,
});
});
if (cards) {
if (cards) {
rateCards.push(cards.in, cards.cached, cards.out);
rateCards.push(cards.in, cards.cached, cards.out);
@@ -202,12 +205,20 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -202,12 +205,20 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const c = route.constraints || {};
const c = route.constraints || {};
const requirements = {
const requirements = {
kind: 'ai', mobility: 'request', latency_class: c.latency_class || 'interactive', objective: c.objective || 'balanced',
kind: 'ai', mobility: 'request', latency_class: c.latency_class || 'interactive', objective: c.objective || 'balanced',
- capabilities: features, units: Number(c.units) || 1, latency_op: operation,
+ capabilities: features.map(f => typeof f === 'string' ? `ai:${f}` : f), units: Number(c.units) || 1, latency_op: operation,
...(c.max_latency_ms != null ? { max_latency_ms: Number(c.max_latency_ms) } : {}),
...(c.max_latency_ms != null ? { max_latency_ms: Number(c.max_latency_ms) } : {}),
...(c.max_cost_usd != null ? { max_cost_usd: Number(c.max_cost_usd) } : {}),
...(c.max_cost_usd != null ? { max_cost_usd: Number(c.max_cost_usd) } : {}),
...(c.trust ? { trust: c.trust } : {}),
...(c.trust ? { trust: c.trust } : {}),
...(c.authority ? { authority: c.authority } : {}),
...(c.authority ? { authority: c.authority } : {}),
};
};
+ for (const offer of offers) {
+ const validation = contracts.validate('platform.resource-offer@1', offer);
+ if (!validation.valid) throw new Error(`invalid platform.resource-offer@1 ${offer.offer_id}: ${JSON.stringify(validation.errors)}`);
+ }
+ // These are SDK planner options, not fields in the workload contract.
+ const { units, latency_op, authority, ...contractRequirements } = requirements;
+ const validation = contracts.validate('platform.workload-requirements@1', contractRequirements);
+ if (!validation.valid) throw new Error(`invalid platform.workload-requirements@1: ${JSON.stringify(validation.errors)}`);
const state = await getRouteState.get(route.key);
const state = await getRouteState.get(route.key);
const current = state && state.current_provider ? offerId(state.current_provider, state.current_model) : null;
const current = state && state.current_provider ? offerId(state.current_provider, state.current_model) : null;
const states = await providerStates(cardsByProvider);
const states = await providerStates(cardsByProvider);
@@ -96,7 +96,21 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open o
@@ -96,7 +96,21 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open o
assert.ok(byProvider.cheap.cost < byProvider.pricey.cost, 'rate cards carry the real prices');
assert.ok(byProvider.cheap.cost < byProvider.pricey.cost, 'rate cards carry the real prices');
// explain is AI's projection of the placement result; that result is a platform.placement-result@1.
// explain is AI's projection of the placement result; that result is a platform.placement-result@1.
- const placed = await h.pool.candidates(r, ['classify'], 'classify');
+ const validated = [];
+ const originalValidate = contracts.validate;
+ let placed;
+ contracts.validate = (name, value) => {
+ const result = originalValidate(name, value);
+ validated.push({ name, value, result });
+ return result;
+ };
+ try { placed = await h.pool.candidates(r, ['classify'], 'classify'); }
+ finally { contracts.validate = originalValidate; }
+ const checkedOffers = validated.filter((x) => x.name === 'platform.resource-offer@1');
+ const checkedRequirements = validated.filter((x) => x.name === 'platform.workload-requirements@1');
+ assert.deepStrictEqual(checkedOffers.map((x) => x.value.offer_id).sort(), placed.placement.candidates.map((x) => x.id).sort(), 'every offer was validated before planning');
+ assert.strictEqual(checkedRequirements.length, 1, 'requirements were validated before planning');
+ for (const x of [...checkedOffers, ...checkedRequirements]) assert.ok(x.result.valid, `${x.name}: ${JSON.stringify(x.result.errors)}`);
const v = contracts.validate('platform.placement-result@1', placed.placement);
const v = contracts.validate('platform.placement-result@1', placed.placement);
assert.ok(v.valid, `platform.placement-result@1: ${JSON.stringify(v.errors)}`);
assert.ok(v.valid, `platform.placement-result@1: ${JSON.stringify(v.errors)}`);
assert.strictEqual(placed.placement.selected, 'cheap:cheap-m');
assert.strictEqual(placed.placement.selected, 'cheap:cheap-m');
@@ -124,6 +138,38 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open o
@@ -124,6 +138,38 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open o
await ok.close(); await flaky.close();
await ok.close(); await flaky.close();
});
});
+t.test('an invalid offer fails closed before reaching the planner', async () => {
+ const r = await h.registry.resolveRoute('test.pool-cheap');
+ const originalListProviders = h.registry.listProviders;
+ const placement = require('openvibe-sdk/placement');
+ const originalPlan = placement.plan;
+ let plannerCalls = 0;
+ h.registry.listProviders = async (...args) => (await originalListProviders.apply(h.registry, args))
+ .map((p) => p.key === 'cheap' ? { ...p, capabilities: ['classify', 42] } : p);
+ placement.plan = () => { plannerCalls++; throw new Error('invalid offer reached planner'); };
+ try {
+ await assert.rejects(h.pool.candidates(r, ['classify'], 'classify'), /invalid platform\.resource-offer@1/);
+ assert.strictEqual(plannerCalls, 0, 'planner was not called with the invalid offer');
+ } finally {
+ h.registry.listProviders = originalListProviders;
+ placement.plan = originalPlan;
+ }
+});
+
+t.test('invalid requirements fail closed before reaching the planner', async () => {
+ const r = await h.registry.resolveRoute('test.pool-cheap');
+ const placement = require('openvibe-sdk/placement');
+ const originalPlan = placement.plan;
+ let plannerCalls = 0;
+ placement.plan = () => { plannerCalls++; throw new Error('invalid requirements reached planner'); };
+ try {
+ await assert.rejects(h.pool.candidates(r, ['classify', 42], 'classify'), /invalid platform\.workload-requirements@1/);
+ assert.strictEqual(plannerCalls, 0, 'planner was not called with invalid requirements');
+ } finally {
+ placement.plan = originalPlan;
+ }
+});
+
t.test('an upstream 429 shifts to another provider without opening the circuit', async () => {
t.test('an upstream 429 shifts to another provider without opening the circuit', async () => {
const ok = await seamServer(() => ({ text: 'steady answer', model: 'steady-m', usage: { input: 3, output: 1 } }));
const ok = await seamServer(() => ({ text: 'steady answer', model: 'steady-m', usage: { input: 3, output: 1 } }));
const busy = await seamServer(() => ({ status: 429, body: { error: { message: 'rate limited' } } }));
const busy = await seamServer(() => ({ status: 429, body: { error: { message: 'rate limited' } } }));