feat(runtime): 端点链接线 —— 策略 → 执行,兜底器全上下文 (#5129) (#5156)

派发步命中分支的 501 换成完整链:匹配 → 策略(E4)→ executeEndpointTarget(E5)
→ 响应映射。cacheTtl 的 Cache-Control 只并进成功答复,错误答复一律不带。

兜底器补三根线:
- 每进程构建一次端点限流注册表(与 server 级限流器同一个 resolveCache);
- 每请求喂完整 EndpointPolicyContext(headers / remoteAddress /
  resolveSessionPrincipalId / limiters / trustProxy / logger),并把
  answer.headers 写到线上(429 的 Retry-After 此前被丢弃);
- 匹配前用 HttpDispatcher.resolveRequestScope(自 dispatch() 原地抽出)解析本
  请求的环境 / 身份 / driver,委派调用带调用方 ExecutionContext 运行。

多租户 host 解析不到环境时该步弃权(不写任何东西),不拿默认 kernel 作答。

现网行为零变更:非空 apis: 在 publish 仍被硬拒(E7 前不撤),整条链结构性不可达。


Claude-Session: https://claude.ai/code/session_01EYGdmvWP1ieZSLqvAW6uyd

Co-authored-by: Claude <noreply@anthropic.com>

Author: os-zhuang

.changeset/endpoint-chain-wiring.md

@@ -0,0 +1,11 @@
+---
+"@objectstack/runtime": minor
+---
+
+端点链接线:声明式 `apis:` 端点的派发步现在跑完整条链 —— 匹配 → 策略(`authRequired` / `rateLimit` / `cacheTtl`)→ 目标执行(`object_operation` 走 `/data` 同一个 `callData`,`flow` 走 automation 服务)。
+
+兜底器(`dispatcher-plugin`)补齐三根一直缺的线:把完整的 `EndpointPolicyContext`(请求头、`remoteAddress`、与 server 级限流器同一个会话查询、每端点限流注册表、`trustProxy`)喂给派发步;把 `answer.headers` **写到线上**(此前 429 的 `Retry-After` 会被丢掉,客户端拿到一个不知道何时重试的 429);并在匹配前解析本请求的环境 / 身份(与 `dispatch()` 同一个 `HttpDispatcher.resolveRequestScope`),使委派调用带着调用方的 `ExecutionContext` 运行,而不是以 system 身份绕过 RLS。
+
+`cacheTtl` 的 `Cache-Control` 只挂在**成功**答复上,任何错误答复都不带它。多租户 host 若无法把请求解析到某个环境,该步**弃权**(不写任何东西,保留传输层原本的 404),而不是拿默认 kernel 的数据来回答。
+
+现网行为零变更:publish / validate 对非空 `apis:` 仍然硬拒(#5040 E7 前不撤),因此本链结构性不可达。

packages/runtime/src/api-endpoint-step.test.ts

@@ -25,6 +25,7 @@ import {
     appEndpointMountPrefix,
     isAppEndpointPath,
     runAppEndpointStep,
+    type AppEndpointExecutionInput,
 } from './api-endpoint-step.js';
 import {
     createEndpointRateLimiterRegistry,
@@ -123,7 +124,7 @@ describe('the step writes nothing unless a declaration owns the request', () =>
     });
 });
 
-describe('a match answers 501 until the executor lands (#5040 E5)', () => {
+describe('a match with no wiring answers an honest 501', () => {
     it('reports NOT_IMPLEMENTED in the declared error envelope', async () => {
         const { service } = matcherFor([TASKS]);
         const answer = await step('/api/v1/apps/showcase/tasks', 'GET', service);
@@ -136,7 +137,7 @@ describe('a match answers 501 until the executor lands (#5040 E5)', () => {
         // Names the endpoint it matched and says plainly that nothing ran —
         // "matched but not executed" must never read as "executed and empty".
         expect(body.error.message).toContain('showcase_tasks');
-        expect(body.error.message).toContain('not enabled');
+        expect(body.error.message).toContain('no wiring');
         expect(String(body.error.hint)).toContain('#5040');
     });
 
@@ -266,3 +267,130 @@ describe('the policy chain runs between the match and the answer', () => {
         expect([...entries.keys()]).toEqual([endpointBucketKey('showcase_tasks', 'principal:usr_7')]);
     });
 });
+
+/**
+ * Execution, wired to the far side of the policy chain (#5040 E5b / #5129).
+ *
+ * The delegation itself is `endpoint-executor.test.ts`'s subject; what is
+ * asserted here is the JOIN — that a passing request reaches the executor with
+ * the request's own coordinates and identity, that a denial never does, and
+ * that `cacheTtl`'s header lands on a success and on nothing else.
+ */
+describe('execution runs on the far side of the policy chain', () => {
+    const OPEN: ApiEndpoint = ApiEndpointSchema.parse({ ...TASKS, name: 'showcase_open', authRequired: false });
+
+    const limiters = () => createEndpointRateLimiterRegistry({ resolveCache: async () => undefined });
+
+    /** Records every delegated `callData` call and answers with a stub result. */
+    function callDataSpy(result: unknown = { object: 'showcase_task', records: [], total: 0 }) {
+        const calls: unknown[][] = [];
+        return {
+            calls,
+            fn: async (...args: unknown[]) => { calls.push(args); return result; },
+        };
+    }
+
+    const wiredStep = (
+        endpoints: ApiEndpoint[],
+        execution: Partial<AppEndpointExecutionInput> & { deps: AppEndpointExecutionInput['deps'] },
+        policy: Partial<EndpointPolicyContext> = {},
+        method = 'GET',
+    ) => runAppEndpointStep({
+        method,
+        path: endpoints[0]!.path,
+        prefix: '/api/v1',
+        metadataService: matcherFor(endpoints).service as never,
+        policy: { limiters: limiters(), ...policy },
+        execution: {
+            request: {
+                method,
+                path: endpoints[0]!.path,
+                query: { limit: '5' },
+                headers: { 'x-caller': 'integration' },
+                body: undefined,
+            },
+            ...execution,
+        },
+    });
+
+    it('delegates a passing request with the request\'s own identity envelope', async () => {
+        const spy = callDataSpy();
+        const executionContext = { userId: 'usr_7', isSystem: false } as never;
+        const answer = await wiredStep([OPEN], {
+            deps: { callData: spy.fn as never },
+            executionContext,
+            environmentId: 'env_1',
+            dataDriver: { driver: true },
+        });
+
+        expect(answer?.status).toBe(200);
+        expect(answer?.body).toEqual({
+            success: true,
+            data: { object: 'showcase_task', records: [], total: 0 },
+            meta: undefined,
+        });
+        // The identity envelope, the driver and the scope ride on the delegated
+        // call — #5040 §4's red line, and the exact thing #4936's dead branch
+        // dropped (it would have read as `system`, RLS bypassed).
+        expect(spy.calls).toEqual([[
+            'query',
+            { object: 'showcase_task', query: { limit: '5' } },
+            { driver: true },
+            'env_1',
+            executionContext,
+        ]]);
+    });
+
+    it('puts the cacheTtl Cache-Control on a SUCCESS answer', async () => {
+        const cached = ApiEndpointSchema.parse({ ...OPEN, name: 'showcase_cached', cacheTtl: 30 });
+        const answer = await wiredStep([cached], { deps: { callData: callDataSpy().fn as never } });
+
+        expect(answer?.status).toBe(200);
+        expect(answer?.headers).toEqual({ 'Cache-Control': 'private, max-age=30' });
+    });
+
+    it('never puts it on an ERROR answer, however the failure arose', async () => {
+        const cached = ApiEndpointSchema.parse({ ...OPEN, name: 'showcase_cached', cacheTtl: 30 });
+        // A delegated pipeline that throws — the executor maps it to a 4xx/5xx
+        // answer, and a client must not be told to reuse a failure for 30s.
+        const answer = await wiredStep([cached], {
+            deps: { callData: async () => { throw { statusCode: 404, message: 'no such object' }; } },
+        });
+
+        expect(answer?.status).toBe(404);
+        expect(answer?.headers).toBeUndefined();
+
+        // Same for a declaration this runtime does not execute (501 from the
+        // executor's own `unsupported` arm, not from the no-wiring branch).
+        const proxied = ApiEndpointSchema.parse({
+            ...OPEN, name: 'showcase_proxy', type: 'proxy', target: 'https://example.invalid', cacheTtl: 30,
+        });
+        const unsupported = await wiredStep([proxied], { deps: { callData: async () => ({}) } });
+        expect(unsupported?.status).toBe(501);
+        expect(unsupported?.headers).toBeUndefined();
+        expect(String((unsupported!.body as { error: { message: string } }).error.message)).toContain('proxy');
+    });
+
+    it('never reaches the executor when a policy denied the request', async () => {
+        const spy = callDataSpy();
+        // `TASKS` keeps the default `authRequired: true`; the caller is anonymous.
+        const answer = await wiredStep([TASKS], { deps: { callData: spy.fn as never } });
+
+        expect(answer?.status).toBe(401);
+        expect(spy.calls, 'the executor ran for a request the policy chain denied').toEqual([]);
+    });
+
+    it('answers an honest 501 when a caller wired policies but no executor', async () => {
+        const answer = await runAppEndpointStep({
+            method: 'GET',
+            path: OPEN.path,
+            prefix: '/api/v1',
+            metadataService: matcherFor([OPEN]).service as never,
+            policy: { limiters: limiters() },
+        });
+        expect(answer?.status).toBe(501);
+        const hint = String((answer!.body as { error: { hint: unknown } }).error.hint);
+        expect(hint).toContain('enforced');
+        expect(hint).toContain('no execution wiring');
+    });
+});