feat: add sourcePolicy to History API values requests

packages/server-api/src/history.ts

@@ -28,9 +28,21 @@ export type AggregateMethod =
 export type ValueList = {
   path: Path
   method: AggregateMethod
-  sourceRef?: SourceRef
+  $source?: SourceRef
 }[]
 
+/**
+ * Controls how history providers should handle multiple sources for the
+ * requested paths.
+ *
+ * - `preferred`: provider default / priority-resolved source handling
+ * - `all`: return values split by source, with `$source` set in `values`
+ *
+ * A `sourceRef` specified on an individual path remains an explicit filter;
+ * `all` only expands paths that do not already specify a source.
+ */
+export type HistorySourcePolicy = 'preferred' | 'all'
+
 /**
  * A row of historical data: first element is timestamp, followed by aggregated values.
  * Values can be primitives, objects (like navigation.position), or null depending on the path.
@@ -104,6 +116,7 @@ export type TimeRangeQueryParams =
 export type ValuesRequestQueryParams = TimeRangeQueryParams & {
   context?: string
   resolution?: number
+  sourcePolicy?: HistorySourcePolicy
 }
 
 export type PathsRequestQueryParams = TimeRangeQueryParams
@@ -258,6 +271,7 @@ export interface PathSpec {
 export type ValuesRequest = TimeRangeParams & {
   context?: Context
   resolution?: number
+  sourcePolicy?: HistorySourcePolicy
   pathSpecs: PathSpec[]
 }
 

packages/server-api/src/typebox/history-schemas.ts

@@ -45,7 +45,13 @@ export const ValuesResponseSchema = Type.Object(
     values: Type.Array(
       Type.Object({
         path: Type.String({ description: 'Signal K path' }),
-        method: Type.String({ description: 'Aggregation method' })
+        method: Type.String({ description: 'Aggregation method' }),
+        $source: Type.Optional(
+          Type.String({
+            description:
+              'Source reference for this value column, present when the values are source-specific'
+          })
+        )
       })
     ),
     data: Type.Array(
@@ -80,7 +86,13 @@ export const PathSpecSchema = Type.Object(
     parameter: Type.Array(Type.String(), {
       description:
         'Additional parameters for the aggregation method (e.g., sample count for sma, alpha for ema)'
-    })
+    }),
+    sourceRef: Type.Optional(
+      Type.String({
+        description:
+          'Optional source reference to filter this path to a single source'
+      })
+    )
   },
   {
     $id: 'PathSpec',

src/api/history/index.ts

@@ -9,6 +9,7 @@ import {
   PathSpec,
   PathsRequest,
   PathsResponse,
+  HistorySourcePolicy,
   TimeRangeParams,
   ValuesRequest,
   ValuesResponse,
@@ -281,6 +282,13 @@ const parseValuesQuery = (query: Record<string, unknown>): ValuesRequest => {
     errors.push('paths parameter is required and must be a string')
   }
 
+  let sourcePolicy: HistorySourcePolicy | undefined
+  try {
+    sourcePolicy = parseSourcePolicy(query.sourcePolicy)
+  } catch (error) {
+    errors.push(error instanceof Error ? error.message : 'Invalid sourcePolicy')
+  }
+
   if (errors.length > 0) {
     throw new Error(`Validation errors: ${errors.join(', ')}`)
   }
@@ -294,12 +302,19 @@ const parseValuesQuery = (query: Record<string, unknown>): ValuesRequest => {
     ...timeRangeParams,
     context,
     resolution,
+    sourcePolicy,
     pathSpecs
   }
   debug.enabled && debug(JSON.stringify(parsed, null, 2))
   return parsed
 }
 
+const parseSourcePolicy = (value: unknown): HistorySourcePolicy | undefined => {
+  if (value === undefined || value === null || value === '') return undefined
+  if (value === 'preferred' || value === 'all') return value
+  throw new Error("sourcePolicy parameter must be 'preferred' or 'all'")
+}
+
 // Maps the single-letter unit suffix in a resolution time expression to seconds.
 const RESOLUTION_UNIT_SECONDS: Record<string, number> = {
   s: 1,

src/api/history/openApi.ts

@@ -150,6 +150,13 @@ historyApiDoc.paths = {
             "Length of data sample time window in seconds or as a time expression ('1s', '1m', '1h', '1d'). If resolution is not specified the server should provide data in a reasonable time resolution, depending on the time range in the request.",
           schema: { type: 'number', format: 'integer' }
         },
+        {
+          name: 'sourcePolicy',
+          in: 'query',
+          description:
+            "Controls how multiple historical sources are handled. With 'all', providers should return values split by source and include $source in the values metadata. A sourceRef specified on an individual path remains an explicit filter; 'all' only expands paths that do not already specify a source.",
+          schema: { type: 'string', enum: ['preferred', 'all'] }
+        },
         { $ref: '#/components/parameters/ProviderIdQuery' }
       ],
       responses: {
@@ -198,7 +205,7 @@ historyApiDoc.paths = {
                           type: 'string',
                           description: 'Aggregation method'
                         },
-                        sourceRef: {
+                        $source: {
                           type: 'string',
                           description:
                             'Source reference, present when the query specified a source filter for this path'

test/history-api.ts

@@ -144,6 +144,37 @@ describe('History API v2', () => {
       body.data.length.should.be.greaterThan(0)
     })
 
+    it('passes sourcePolicy=all to the provider', async function () {
+      const res = await fetch(
+        `${api}/history/values?paths=navigation.speedOverGround&from=${FROM}&to=${TO}&resolution=60&sourcePolicy=all`
+      )
+      res.status.should.equal(200)
+      const body = await res.json()
+      assertSchema(ValuesResponseSchema, body, 'ValuesResponse')
+      body.values.should.deep.equal([
+        {
+          path: 'navigation.speedOverGround',
+          method: 'average',
+          $source: 'source.a'
+        },
+        {
+          path: 'navigation.speedOverGround',
+          method: 'average',
+          $source: 'source.b'
+        }
+      ])
+      body.data.should.deep.equal([['2025-01-01T12:00:00Z', 1.2, 2.4]])
+    })
+
+    it('returns 400 for invalid sourcePolicy', async function () {
+      const res = await fetch(
+        `${api}/history/values?paths=navigation.position&from=${FROM}&to=${TO}&sourcePolicy=unknown`
+      )
+      res.status.should.equal(400)
+      const body = await res.json()
+      body.error.should.contain('sourcePolicy')
+    })
+
     it('returns paths from the provider', async function () {
       const res = await fetch(`${api}/history/paths?from=${FROM}&to=${TO}`)
       res.status.should.equal(200)