Load the version from shared context if set

packages/world/package.json

@@ -23,6 +23,7 @@
     "dist"
   ],
   "dependencies": {
+    "@platformatic/globals": "^3.57.0",
     "cbor-x": "1.6.4",
     "undici": "^8.0.0"
   },

packages/world/src/index.ts

@@ -1,6 +1,6 @@
 import { readFileSync } from 'node:fs'
 import { join } from 'node:path'
-import { Agent } from 'undici'
+import { getSharedContext } from '@platformatic/globals'
 import type { World } from '@workflow/world'
 import { SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT } from '@workflow/world'
 import { HttpClient } from './lib/client.ts'
@@ -56,6 +56,21 @@ export interface CreateWorldOptions {
   deploymentVersion: string
 }
 
+// Read the deployment version from the watt runtime shared context, if something
+// pushed one there. getSharedContext returns { get, update } inside a runtime, or
+// undefined off-runtime with throwOnMissing:false -- a safe no-op standalone.
+async function versionFromSharedContext (): Promise<string | undefined> {
+  try {
+    const shared = getSharedContext({ throwOnMissing: false }) as
+      { get?: () => unknown } | undefined
+    if (!shared?.get) return undefined
+    const ctx = await shared.get() as { deploymentVersion?: string } | undefined
+    return ctx?.deploymentVersion
+  } catch {
+    return undefined
+  }
+}
+
 function saPath (file: string): string {
   const base = process.env.PLT_WORLD_SA_PATH || '/var/run/secrets/kubernetes.io/serviceaccount'
   return `${base}/${file}`
@@ -70,31 +85,6 @@ function isRunningInK8s (): boolean {
   }
 }
 
-async function readVersionFromK8sApi (): Promise<string | undefined> {
-  try {
-    const token = readFileSync(saPath('token'), 'utf8').trim()
-    const namespace = readFileSync(saPath('namespace'), 'utf8').trim()
-    const podName = process.env.HOSTNAME
-    if (!podName) return undefined
-
-    const ca = readFileSync(saPath('ca.crt'))
-    const dispatcher = new Agent({ connect: { ca } })
-    const res = await fetch(
-      `https://kubernetes.default.svc/api/v1/namespaces/${namespace}/pods/${podName}`,
-      {
-        headers: { Authorization: `Bearer ${token}` },
-        dispatcher,
-      } as RequestInit
-    )
-
-    if (!res.ok) return undefined
-    const pod = await res.json() as { metadata?: { labels?: Record<string, string> } }
-    return pod?.metadata?.labels?.['plt.dev/version'] || undefined
-  } catch {
-    return undefined
-  }
-}
-
 function readAppName (): string {
   try {
     const pkg = JSON.parse(readFileSync(join(process.cwd(), 'package.json'), 'utf8'))
@@ -111,19 +101,34 @@ export function createWorld (options?: Partial<CreateWorldOptions>): World {
   }
 
   const appId = options?.appId || process.env.PLT_WORLD_APP_ID || readAppName()
-  const explicitVersion = options?.deploymentVersion || process.env.PLT_WORLD_DEPLOYMENT_VERSION
-  const config = { serviceUrl, appId, deploymentVersion: explicitVersion || 'local' }
-  const world = createPlatformaticWorld(config)
+  const explicitVersion = options?.deploymentVersion ||
+    process.env.PLT_WORLD_DEPLOYMENT_VERSION ||
+    process.env.PLT_DEPLOYMENT_VERSION
+  // Version comes from the environment first: options, PLT_WORLD_DEPLOYMENT_VERSION,
+  // or PLT_DEPLOYMENT_VERSION. No K8s API read.
+  const config: PlatformaticWorldConfig = {
+    serviceUrl,
+    appId,
+    deploymentVersion: explicitVersion || 'local',
+    // In K8s ICC assigns the version, so a 'local' stamp means "not resolved yet" and
+    // must not be used to enqueue (see queue.ts). Standalone/local dev keeps 'local'.
+    requireResolvedVersion: isRunningInK8s(),
+  }
 
-  if (!explicitVersion && isRunningInK8s()) {
-    const originalStart = world.start!
-    world.start = async function () {
-      const version = await readVersionFromK8sApi()
+  // No explicit version: start at 'local'. When running inside a watt runtime, the
+  // version can be provided later via the shared context, so refresh before each
+  // queue read and latch it -- the pod starts stamping the real version with no
+  // restart. Off-runtime (standalone) this is a no-op.
+  if (!explicitVersion) {
+    config.refreshDeploymentVersion = async () => {
+      if (config.deploymentVersion !== 'local') return
+      const version = await versionFromSharedContext()
       if (version) config.deploymentVersion = version
-      return originalStart.call(this)
     }
   }
 
+  const world = createPlatformaticWorld(config)
+
   return world
 }
 

packages/world/src/lib/queue.ts

@@ -5,6 +5,13 @@ import { decode } from 'cbor-x'
 
 export interface QueueConfig {
   deploymentVersion: string
+  // Optional late resolver: when the version was not known at startup
+  refreshDeploymentVersion?: () => Promise<void>
+  // When true (K8s, where ICC assigns the version), 'local' means "not resolved yet",
+  // so enqueuing is refused instead of stamping 'local' -- otherwise the run pins to
+  // 'local' and its messages never route to the ICC-registered (real-version)
+  // handlers. In standalone/local dev this is false and 'local' is a valid version.
+  requireResolvedVersion?: boolean
 }
 
 interface EnqueueEnvelope {
@@ -28,10 +35,19 @@ async function readRequestBody (req: Request): Promise<Buffer> {
 
 export function createQueue (client: HttpClient, config: QueueConfig) {
   const queue = async (queueName: ValidQueueName, message: QueuePayload, opts?: QueueOptions): Promise<{ messageId: MessageId | null }> => {
+    // Pick up a version that arrived after startup (ICC recovery) before stamping.
+    await config.refreshDeploymentVersion?.()
+    const deploymentId = opts?.deploymentId ?? config.deploymentVersion
+    // In K8s the version is assigned by ICC, so 'local' means "not resolved yet".
+    // Refuse rather than pin the run to 'local' (its messages would never route to
+    // the real-version handlers). The caller retries until the version lands.
+    if (config.requireResolvedVersion && deploymentId === 'local') {
+      throw new Error('World deployment version not resolved yet (ICC unreachable); retry')
+    }
     const envelope: EnqueueEnvelope = {
       queueName,
       message,
-      deploymentId: opts?.deploymentId ?? config.deploymentVersion,
+      deploymentId,
       idempotencyKey: opts?.idempotencyKey,
       delaySeconds: opts?.delaySeconds,
     }
@@ -80,6 +96,7 @@ export function createQueue (client: HttpClient, config: QueueConfig) {
   }
 
   const getDeploymentId = async (): Promise<string> => {
+    await config.refreshDeploymentVersion?.()
     return config.deploymentVersion
   }
 

packages/world/test/world.test.ts

@@ -67,9 +67,11 @@ test('reads from env vars', async () => {
 test('defaults deploymentVersion to local', async () => {
   const originalUrl = process.env.PLT_WORLD_SERVICE_URL
   const originalVersion = process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  const originalGeneralVersion = process.env.PLT_DEPLOYMENT_VERSION
 
   process.env.PLT_WORLD_SERVICE_URL = 'http://localhost:9999'
   delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  delete process.env.PLT_DEPLOYMENT_VERSION
 
   try {
     const world = createWorld()
@@ -81,6 +83,113 @@ test('defaults deploymentVersion to local', async () => {
     else delete process.env.PLT_WORLD_SERVICE_URL
     if (originalVersion) process.env.PLT_WORLD_DEPLOYMENT_VERSION = originalVersion
     else delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+    if (originalGeneralVersion) process.env.PLT_DEPLOYMENT_VERSION = originalGeneralVersion
+    else delete process.env.PLT_DEPLOYMENT_VERSION
+  }
+})
+
+test('falls back to PLT_DEPLOYMENT_VERSION when PLT_WORLD_DEPLOYMENT_VERSION is unset', async () => {
+  const originalUrl = process.env.PLT_WORLD_SERVICE_URL
+  const originalWorldVersion = process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  const originalVersion = process.env.PLT_DEPLOYMENT_VERSION
+
+  process.env.PLT_WORLD_SERVICE_URL = 'http://localhost:9999'
+  delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  process.env.PLT_DEPLOYMENT_VERSION = 'img-1.4.2'
+
+  try {
+    const world = createWorld()
+    const deploymentId = await world.getDeploymentId()
+    assert.equal(deploymentId, 'img-1.4.2')
+    await world.close()
+  } finally {
+    if (originalUrl) process.env.PLT_WORLD_SERVICE_URL = originalUrl
+    else delete process.env.PLT_WORLD_SERVICE_URL
+    if (originalWorldVersion) process.env.PLT_WORLD_DEPLOYMENT_VERSION = originalWorldVersion
+    else delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+    if (originalVersion) process.env.PLT_DEPLOYMENT_VERSION = originalVersion
+    else delete process.env.PLT_DEPLOYMENT_VERSION
+  }
+})
+
+test('starts at local and picks up the version pushed to the shared context later', async () => {
+  const originalUrl = process.env.PLT_WORLD_SERVICE_URL
+  const originalWorldVersion = process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  const originalVersion = process.env.PLT_DEPLOYMENT_VERSION
+  const originalGlobal = (globalThis as any).platformatic
+
+  process.env.PLT_WORLD_SERVICE_URL = 'http://localhost:9999'
+  delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  delete process.env.PLT_DEPLOYMENT_VERSION
+
+  // Simulate the watt runtime shared context; starts with no version.
+  let sharedCtx: { deploymentVersion?: string } = {}
+  ;(globalThis as any).platformatic = { sharedContext: { get: () => sharedCtx } }
+
+  try {
+    const world = createWorld()
+    // No env version and nothing in the shared context yet -> local.
+    assert.equal(await world.getDeploymentId(), 'local')
+
+    // Version arrives later (e.g. watt-extra pushes it once ICC resolves it).
+    sharedCtx = { deploymentVersion: 'v9' }
+    assert.equal(await world.getDeploymentId(), 'v9')
+
+    // Latched: once resolved it stops consulting the shared context.
+    sharedCtx = {}
+    assert.equal(await world.getDeploymentId(), 'v9')
+
+    await world.close()
+  } finally {
+    if (originalUrl) process.env.PLT_WORLD_SERVICE_URL = originalUrl
+    else delete process.env.PLT_WORLD_SERVICE_URL
+    if (originalWorldVersion) process.env.PLT_WORLD_DEPLOYMENT_VERSION = originalWorldVersion
+    else delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+    if (originalVersion) process.env.PLT_DEPLOYMENT_VERSION = originalVersion
+    else delete process.env.PLT_DEPLOYMENT_VERSION
+    ;(globalThis as any).platformatic = originalGlobal
+  }
+})
+
+test('refuses to enqueue while the version is unresolved (local) in K8s', async () => {
+  const originalUrl = process.env.PLT_WORLD_SERVICE_URL
+  const originalSaPath = process.env.PLT_WORLD_SA_PATH
+  const originalWorldVersion = process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  const originalVersion = process.env.PLT_DEPLOYMENT_VERSION
+  const originalGlobal = (globalThis as any).platformatic
+
+  // Fake SA mount so isRunningInK8s() is true.
+  const fakeSaDir = join(tmpdir(), `plt-world-k8s-block-${process.pid}`)
+  mkdirSync(fakeSaDir, { recursive: true })
+  writeFileSync(join(fakeSaDir, 'token'), 'fake-token')
+
+  process.env.PLT_WORLD_SERVICE_URL = 'http://localhost:9999'
+  process.env.PLT_WORLD_SA_PATH = fakeSaDir
+  delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+  delete process.env.PLT_DEPLOYMENT_VERSION
+  ;(globalThis as any).platformatic = undefined // no shared context -> stays local
+
+  try {
+    const world = createWorld()
+    assert.equal(await world.getDeploymentId(), 'local')
+    // Enqueuing is refused so the run is not pinned to 'local' (its messages would
+    // never route to the ICC-registered real-version handlers). Caller retries.
+    await assert.rejects(
+      () => world.queue('my-queue' as any, {} as any),
+      /not resolved/
+    )
+    await world.close()
+  } finally {
+    if (originalUrl) process.env.PLT_WORLD_SERVICE_URL = originalUrl
+    else delete process.env.PLT_WORLD_SERVICE_URL
+    if (originalSaPath) process.env.PLT_WORLD_SA_PATH = originalSaPath
+    else delete process.env.PLT_WORLD_SA_PATH
+    if (originalWorldVersion) process.env.PLT_WORLD_DEPLOYMENT_VERSION = originalWorldVersion
+    else delete process.env.PLT_WORLD_DEPLOYMENT_VERSION
+    if (originalVersion) process.env.PLT_DEPLOYMENT_VERSION = originalVersion
+    else delete process.env.PLT_DEPLOYMENT_VERSION
+    ;(globalThis as any).platformatic = originalGlobal
+    rmSync(fakeSaDir, { recursive: true, force: true })
   }
 })
 
@@ -114,7 +223,7 @@ test('falls back to local when not in K8s and no env var', async () => {
   }
 })
 
-test('env var takes precedence over K8s API lookup', async () => {
+test('reads the deployment version from PLT_WORLD_DEPLOYMENT_VERSION', async () => {
   const originalUrl = process.env.PLT_WORLD_SERVICE_URL
   const originalVersion = process.env.PLT_WORLD_DEPLOYMENT_VERSION