From ff5410e1bfbcba0e594b757900f299a053662c79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Wed, 2 Sep 2026 12:30:38 +0200 Subject: [PATCH 1/4] Add Scala agent reflection APIs --- docs/src/content/next/how-to-guides.mdx | 2 +- docs/src/content/next/how-to-guides/scala.mdx | 1 + .../content/next/how-to-guides/scala/_meta.js | 1 + .../golem-call-agent-reflectively-scala.mdx | 95 +++ .../SKILL.md | 100 +++ .../scenarios/rpc-5-runtime-reflection.yaml | 58 ++ .../golem/reflection/CallerCodecClient.scala | 161 +++++ .../scala/golem/reflection/Reflection.scala | 435 +++++++++++++ .../scala/golem/reflection/SchemaRef.scala | 588 ++++++++++++++++++ .../golem/runtime/rpc/host/WasmRpcApi.scala | 2 +- .../golem/reflection/SchemaRefSpec.scala | 102 +++ 11 files changed, 1543 insertions(+), 2 deletions(-) create mode 100644 docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx create mode 100644 golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md create mode 100644 sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala create mode 100644 sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala create mode 100644 sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala create mode 100644 sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala diff --git a/docs/src/content/next/how-to-guides.mdx b/docs/src/content/next/how-to-guides.mdx index 3f7aea9239..68b3da1f15 100644 --- a/docs/src/content/next/how-to-guides.mdx +++ b/docs/src/content/next/how-to-guides.mdx @@ -8,6 +8,6 @@ Practical, step-by-step guides for building with Golem. Each guide covers a spec - + diff --git a/docs/src/content/next/how-to-guides/scala.mdx b/docs/src/content/next/how-to-guides/scala.mdx index b8fd483c66..25ff65ec16 100644 --- a/docs/src/content/next/how-to-guides/scala.mdx +++ b/docs/src/content/next/how-to-guides/scala.mdx @@ -15,6 +15,7 @@ Guides specific to developing Golem agents in Scala. + diff --git a/docs/src/content/next/how-to-guides/scala/_meta.js b/docs/src/content/next/how-to-guides/scala/_meta.js index c9766a2535..e5ce891992 100644 --- a/docs/src/content/next/how-to-guides/scala/_meta.js +++ b/docs/src/content/next/how-to-guides/scala/_meta.js @@ -9,6 +9,7 @@ export default { "golem-annotate-agent-scala": "Annotating Agent Methods (Scala)", "golem-atomic-block-scala": "Atomic Blocks and Durability Controls (Scala)", "golem-call-from-external-scala": "Calling Agents from External Applications (Scala)", + "golem-call-agent-reflectively-scala": "Calling Agents with Runtime Reflection (Scala)", "golem-call-another-agent-scala": "Calling Another Agent (Scala)", "golem-configure-durability-scala": "Configuring Agent Durability (Scala)", "golem-add-cors-scala": "Configuring CORS for Scala HTTP Endpoints", diff --git a/docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx b/docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx new file mode 100644 index 0000000000..03a1e36954 --- /dev/null +++ b/docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx @@ -0,0 +1,95 @@ +# Calling Agents with Runtime Reflection (Scala) + +Use generated clients when the complete target definition is available at +compile time. Otherwise choose one authority and invocation path explicitly: +caller-authored codecs, runtime-reflected schemas, or direct schema-free +`SchemaValue` calls. + +## Discover and inspect schemas + +```scala +import golem.reflection.Reflection + +val target = Reflection.getAgentType("CounterAgent").flatMap( + _.toRight(golem.reflection.GolemReflectError.Discovery("CounterAgent is unavailable")) +) +val method = target.flatMap( + _.method("add").toRight(golem.reflection.GolemReflectError.Discovery("add is unavailable")) +) +``` + +An `AgentType` exposes its component ID, lifecycle mode, constructor +`SchemaRef`, and method `SchemaRef`s. `SchemaRef` validates `SchemaValue`, packs +and unpacks canonical `zio.blocks.schema.json.Json`, and renders JSON Schema. + +## Use the three reflected/value invocation paths + +JSON convenience automatically packs and unpacks: + +```scala +val invocation = client.method("add").flatMap { add => + // The returned Future contains either a reflection error or JSON invocation. + Right(add.invokeJson(Json.Object("by" -> Json.Number(BigDecimal(5))))) +} +``` + +For explicit reflected packing, call `method.definition.input.packJson`, then +`invokeValue`; after awaiting, call the output `SchemaRef.unpackJson`. + +For direct values, bind `agentId.dynamicClient` and manually construct the +positional record: + +```scala +import golem.schema.SchemaValue + +val call = agentId.dynamicClient.map( + _.method("add").invokeValue( + SchemaValue.RecordValue(List(SchemaValue.U32Value(5))) + ) +) +``` + +Direct clients never discover or validate schemas. Constructor and method +record fields must be packed in declaration order; the runtime authoritatively +accepts or rejects the attempt. + +All reflected and direct methods support awaited, trigger, and scheduled calls +through `invokeValue`, `triggerValue`, and `scheduleValue`. + +## Define a caller-codec typed contract + +`AgentClientDefinition` is discovery-free. `InputRecordCodec` and +`OutputCodec` are the caller's schema authority: + +```scala +import golem.reflection._ +import golem.runtime.{InputRecordCodec, OutputCodec} + +val contract = AgentClientDefinition( + componentId, + "CounterAgent", + InputRecordCodec.single[String]("name") +) +val add = contract.method( + "add", + InputRecordCodec.single[Int]("by"), + OutputCodec.single[Int] +) + +val counter = contract.client.get("main") +val result = counter.map(_.method(add).invoke(5)) +``` + +## Lifecycle attempts + +- Use a supplied `AgentId` directly or inspect it with `parts`. +- Use `AgentId.create` for schema-free durable, known-phantom, or newly + generated phantom identities. +- Reflected and caller-codec factories provide `get`, `getPhantom`, and + `newPhantom`. +- Use `DynamicAgentClient.ephemeral(componentId, typeName, constructorValue)` + for a raw ephemeral invocation address. + +An ephemeral address has no guaranteed reusable pre-invocation identity. The +final identity comes from invocation metadata and must not be treated as a +resumable durable identity. diff --git a/golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md b/golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md new file mode 100644 index 0000000000..35f3e1fab6 --- /dev/null +++ b/golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md @@ -0,0 +1,100 @@ +--- +name: golem-call-agent-reflectively-scala +description: "Discovering and calling Golem agents through runtime reflection in Scala. Use when agent types or methods are selected dynamically, schemas must be inspected at runtime, caller-owned codecs are needed, or SchemaValue calls must avoid discovery." +--- + +# Calling Agents with Runtime Reflection (Scala) + +Use generated clients when the complete target definition is available at +compile time. Otherwise choose one authority and invocation path explicitly: +caller-authored codecs, runtime-reflected schemas, or direct schema-free +`SchemaValue` calls. + +## Discover and inspect schemas + +```scala +import golem.reflection.Reflection + +val target = Reflection.getAgentType("CounterAgent").flatMap( + _.toRight(golem.reflection.GolemReflectError.Discovery("CounterAgent is unavailable")) +) +val method = target.flatMap( + _.method("add").toRight(golem.reflection.GolemReflectError.Discovery("add is unavailable")) +) +``` + +An `AgentType` exposes its component ID, lifecycle mode, constructor +`SchemaRef`, and method `SchemaRef`s. `SchemaRef` validates `SchemaValue`, packs +and unpacks canonical `zio.blocks.schema.json.Json`, and renders JSON Schema. + +## Use the three reflected/value invocation paths + +JSON convenience automatically packs and unpacks: + +```scala +val invocation = client.method("add").flatMap { add => + // The returned Future contains either a reflection error or JSON invocation. + Right(add.invokeJson(Json.Object("by" -> Json.Number(BigDecimal(5))))) +} +``` + +For explicit reflected packing, call `method.definition.input.packJson`, then +`invokeValue`; after awaiting, call the output `SchemaRef.unpackJson`. + +For direct values, bind `agentId.dynamicClient` and manually construct the +positional record: + +```scala +import golem.schema.SchemaValue + +val call = agentId.dynamicClient.map( + _.method("add").invokeValue( + SchemaValue.RecordValue(List(SchemaValue.U32Value(5))) + ) +) +``` + +Direct clients never discover or validate schemas. Constructor and method +record fields must be packed in declaration order; the runtime authoritatively +accepts or rejects the attempt. + +All reflected and direct methods support awaited, trigger, and scheduled calls +through `invokeValue`, `triggerValue`, and `scheduleValue`. + +## Define a caller-codec typed contract + +`AgentClientDefinition` is discovery-free. `InputRecordCodec` and +`OutputCodec` are the caller's schema authority: + +```scala +import golem.reflection._ +import golem.runtime.{InputRecordCodec, OutputCodec} + +val contract = AgentClientDefinition( + componentId, + "CounterAgent", + InputRecordCodec.single[String]("name") +) +val add = contract.method( + "add", + InputRecordCodec.single[Int]("by"), + OutputCodec.single[Int] +) + +val counter = contract.client.get("main") +val result = counter.map(_.method(add).invoke(5)) +``` + +## Lifecycle attempts + +- Use a supplied `AgentId` directly or inspect it with `parts`. +- Use `AgentId.create` for schema-free durable, known-phantom, or newly + generated phantom identities. +- Reflected and caller-codec factories provide `get`, `getPhantom`, and + `newPhantom`. +- Use `DynamicAgentClient.ephemeral(componentId, typeName, constructorValue)` + for a raw ephemeral invocation address. + +An ephemeral address has no guaranteed reusable pre-invocation identity. The +final identity comes from invocation metadata and must not be treated as a +resumable durable identity. diff --git a/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml b/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml index 3749ae236f..9a1abc75a3 100644 --- a/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml +++ b/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml @@ -134,3 +134,61 @@ steps: equals: "reflected value" - path: "$.direct_value" equals: "direct value" + + - id: "create-scala-project" + only_if: + language: "scala" + create_project: + name: test-app + verify: + build: true + + - id: "add-scala-reflected-rpc" + only_if: + language: "scala" + prompt: > + In the Scala test-app project, add a durable `ReflectionTarget` agent + identified by a string `name`, with an `echo` method accepting and + returning a string. Add a durable `ReflectionCaller` whose async `run` + method discovers `ReflectionTarget`, inspects `echo`, and invokes it + through the JSON convenience path with `hello`. + + Also define a discovery-free caller-codec contract using + `InputRecordCodec` and `OutputCodec`, and invoke `echo` with + `caller codec`. Explicitly pack JSON through the reflected method + `SchemaRef`, invoke its value API, await the result, and unpack it as + `reflected value`. Finally bind a dynamic client to the same AgentId, + manually construct the positional `SchemaValue.RecordValue`, and invoke + `echo` with `direct value` without schemas. + + Return a schema-derived record containing `listed`, `typeName`, + `methodName`, `jsonValue`, `callerCodec`, `reflectedValue`, and + `directValue`. Make sure the project builds successfully. + expectedSkills: + - "golem-call-agent-reflectively-scala" + verify: + build: true + deploy: true + + - id: "verify-scala-reflected-rpc" + only_if: + language: "scala" + invoke_json: + agent: 'ReflectionCaller("main")' + method: "run" + expect: + result_json: + - path: "$.listed" + equals: true + - path: "$.typeName" + equals: "ReflectionTarget" + - path: "$.methodName" + equals: "echo" + - path: "$.jsonValue" + equals: "hello" + - path: "$.callerCodec" + equals: "caller codec" + - path: "$.reflectedValue" + equals: "reflected value" + - path: "$.directValue" + equals: "direct value" diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala new file mode 100644 index 0000000000..1e29e67c24 --- /dev/null +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala @@ -0,0 +1,161 @@ +/* + * Copyright 2024-2026 Golem Cloud + * + * Licensed under the Golem Source License v1.1 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://license.golem.cloud/LICENSE + */ + +package golem.reflection + +import golem.runtime.{InputRecordCodec, OutputCodec, OutputMetadata} +import golem.schema.SchemaValue +import golem.{Datetime, Uuid} + +import scala.concurrent.Future +import scala.scalajs.concurrent.JSExecutionContext.Implicits.queue +import scala.util.control.NonFatal + +/** A discovery-free, caller-authored typed agent contract. */ +final class AgentClientDefinition[Constructor] private ( + val componentId: ComponentId, + val name: String, + val mode: AgentMode, + val constructor: InputRecordCodec[Constructor] +) { + val client: CallerCodecClientFactory[Constructor] = new CallerCodecClientFactory(this) + + def method[Input, Output]( + name: String, + input: InputRecordCodec[Input], + output: OutputCodec[Output] + ): CallerCodecMethod[Input, Output] = + CallerCodecMethod(name, input, output) +} + +object AgentClientDefinition { + def apply[Constructor]( + componentId: ComponentId, + name: String, + constructor: InputRecordCodec[Constructor], + mode: AgentMode = AgentMode.Durable + ): AgentClientDefinition[Constructor] = + new AgentClientDefinition(componentId, name, mode, constructor) +} + +final case class CallerCodecMethod[Input, Output]( + name: String, + input: InputRecordCodec[Input], + output: OutputCodec[Output] +) + +final case class CallerCodecPhantomClient[Constructor]( + agentId: AgentId, + phantomId: Uuid, + client: CallerCodecAgentClient[Constructor] +) + +final class CallerCodecClientFactory[Constructor] private[reflection] ( + definition: AgentClientDefinition[Constructor] +) { + def get(input: Constructor): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = + requireDurable("get").flatMap(_ => create(input, None)) + + def getPhantom(input: Constructor, phantomId: Uuid): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = + create(input, Some(phantomId)) + + def newPhantom( + input: Constructor + ): Either[GolemReflectError, Either[CallerCodecAgentClient[Constructor], CallerCodecPhantomClient[Constructor]]] = + if (definition.mode == AgentMode.Ephemeral) create(input, None).map(Left(_)) + else { + val phantom = Uuid.random() + for { + constructor <- encodeConstructor(input) + id <- AgentId.create(definition.componentId, definition.name, constructor, Some(phantom)) + client <- createValue(constructor, Some(phantom)) + } yield Right(CallerCodecPhantomClient(id, phantom, client)) + } + + private def create( + input: Constructor, + phantomId: Option[Uuid] + ): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = + encodeConstructor(input).flatMap(createValue(_, phantomId)) + + private def createValue( + constructor: SchemaValue, + phantomId: Option[Uuid] + ): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = + Transport + .create(definition.componentId, definition.name, constructor, phantomId) + .map(new CallerCodecAgentClient(definition, _)) + + private def encodeConstructor(input: Constructor): Either[GolemReflectError, SchemaValue] = + try Right(definition.constructor.toValue(input)) + catch { case NonFatal(error) => Left(GolemReflectError.SchemaEncode(error.getMessage)) } + + private def requireDurable(operation: String): Either[GolemReflectError, Unit] = + Either.cond( + definition.mode == AgentMode.Durable, + (), + GolemReflectError.Identity(s"$operation is not available for ephemeral agent types") + ) +} + +final class CallerCodecAgentClient[Constructor] private[reflection] ( + definition: AgentClientDefinition[Constructor], + transport: Transport +) { + def method[Input, Output](definition: CallerCodecMethod[Input, Output]): CallerCodecBoundMethod[Input, Output] = + new CallerCodecBoundMethod(definition, transport) +} + +final class CallerCodecBoundMethod[Input, Output] private[reflection] ( + definition: CallerCodecMethod[Input, Output], + transport: Transport +) { + def invoke(input: Input): Future[Either[GolemReflectError, TypedInvocation[Output]]] = + encodeInput(input) match { + case Left(error) => Future.successful(Left(error)) + case Right(value) => + transport + .invokeAndAwait(definition.name, value) + .map(_.flatMap { invocation => + decodeOutput(invocation.value).map(output => TypedInvocation(invocation.metadata, output)) + }) + } + + def trigger(input: Input): Either[GolemReflectError, InvocationMetadata] = + encodeInput(input).flatMap(transport.trigger(definition.name, _)) + + def schedule(at: Datetime, input: Input): Either[GolemReflectError, ScheduledInvocation] = + encodeInput(input).flatMap(transport.schedule(at, definition.name, _)) + + private def encodeInput(input: Input): Either[GolemReflectError, SchemaValue] = + try Right(definition.input.toValue(input)) + catch { case NonFatal(error) => Left(GolemReflectError.SchemaEncode(error.getMessage)) } + + private def decodeOutput(value: Option[SchemaValue]): Either[GolemReflectError, Output] = + definition.output.metadata match { + case OutputMetadata.Unit => + Either.cond( + value.isEmpty, + ().asInstanceOf[Output], + GolemReflectError.SchemaDecode("unit method returned a value") + ) + case OutputMetadata.Single(_) => + value + .toRight(GolemReflectError.SchemaDecode("single-output method returned no value")) + .flatMap(schemaValue => + definition.output.from.get + .fromValue(schemaValue) + .left + .map(error => GolemReflectError.SchemaDecode(error.message)) + ) + } +} + +final case class TypedInvocation[+A](metadata: InvocationMetadata, value: A) diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala new file mode 100644 index 0000000000..b71a1187a7 --- /dev/null +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala @@ -0,0 +1,435 @@ +/* + * Copyright 2024-2026 Golem Cloud + * + * Licensed under the Golem Source License v1.1 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://license.golem.cloud/LICENSE + */ + +package golem.reflection + +import golem.host.SchemaWireInterop +import golem.host.js.{JsComponentId, JsUuid} +import golem.host.js.schema.{ + JsInputSchema, + JsNamedField, + JsOutputSchema, + JsSchemaGraph, + JsSchemaValueTree, + JsUuid => JsSchemaUuid +} +import golem.runtime.rpc.host.{AgentHostApi, WasmRpcApi} +import golem.runtime.rpc.{CancellationToken, InvocationReceipt} +import golem.schema._ +import golem.schema.SchemaTypeBody.RecordType +import golem.schema.validation.ValueValidation +import golem.schema.wire.SchemaWire +import golem.{Datetime, FutureInterop, Uuid} +import zio.blocks.schema.json.Json + +import scala.concurrent.Future +import scala.scalajs.js +import scala.scalajs.js.JSConverters._ +import scala.scalajs.concurrent.JSExecutionContext.Implicits.queue +import scala.util.control.NonFatal + +import ReflectionInternals._ + +sealed trait AgentMode extends Product with Serializable +object AgentMode { + case object Durable extends AgentMode + case object Ephemeral extends AgentMode +} + +final case class ComponentId(uuid: Uuid) { + private[golem] def toJs: JsComponentId = + JsComponentId(JsUuid(js.BigInt(uuid.highBits.toString), js.BigInt(uuid.lowBits.toString))) +} + +object ComponentId { + private[golem] def fromJs(value: JsComponentId): ComponentId = + ComponentId(Uuid(BigInt(value.uuid.highBits.toString), BigInt(value.uuid.lowBits.toString))) +} + +final case class AgentId(componentId: ComponentId, value: String) { + def parts: Either[GolemReflectError, AgentIdParts] = AgentId.parse(this) + def dynamicClient: Either[GolemReflectError, DynamicAgentClient] = DynamicAgentClient.fromAgentId(this) +} + +final case class AgentIdParts(typeName: String, constructorValue: SchemaValue, phantomId: Option[Uuid]) + +object AgentId { + def create( + componentId: ComponentId, + typeName: String, + constructorValue: SchemaValue, + phantomId: Option[Uuid] = None + ): Either[GolemReflectError, AgentId] = + encode(constructorValue).flatMap(payload => + AgentHostApi + .makeAgentId(typeName, payload, phantomId) + .left + .map(GolemReflectError.Identity) + .map(AgentId(componentId, _)) + ) + + def parse(agentId: AgentId): Either[GolemReflectError, AgentIdParts] = + AgentHostApi + .parseAgentId(agentId.value) + .left + .map(GolemReflectError.Identity) + .flatMap { parts => + try + Right( + AgentIdParts( + parts.agentTypeName, + SchemaWire.schemaValueFromWit(SchemaWireInterop.valueTreeFromJs(parts.payload.value)), + parts.phantom + ) + ) + catch { case NonFatal(error) => Left(GolemReflectError.SchemaDecode(error.getMessage)) } + } +} + +sealed trait GolemReflectError extends Product with Serializable { + def message: String + override def toString: String = message +} + +object GolemReflectError { + final case class Discovery(message: String) extends GolemReflectError + final case class Identity(message: String) extends GolemReflectError + final case class SchemaEncode(message: String) extends GolemReflectError + final case class SchemaDecode(message: String) extends GolemReflectError + final case class Validation(message: String) extends GolemReflectError + final case class Remote(message: String) extends GolemReflectError +} + +final case class AgentMethod( + name: String, + description: String, + promptHint: Option[String], + input: SchemaRef, + output: Option[SchemaRef] +) + +final class AgentType private[reflection] ( + val name: String, + val description: String, + val sourceLanguage: String, + val mode: AgentMode, + val implementedBy: ComponentId, + val constructorInput: SchemaRef, + val methods: List[AgentMethod] +) { + val client: ReflectedAgentClientFactory = new ReflectedAgentClientFactory(this) + + def method(name: String): Option[AgentMethod] = methods.find(_.name == name) + + def agentId(input: Json, phantomId: Option[Uuid] = None): Either[GolemReflectError, AgentId] = + constructorInput + .packJson(input) + .left + .map(error => GolemReflectError.Validation(error.message)) + .flatMap(agentIdValue(_, phantomId)) + + def agentIdValue(input: SchemaValue, phantomId: Option[Uuid] = None): Either[GolemReflectError, AgentId] = + validate(constructorInput, input).flatMap(_ => AgentId.create(implementedBy, name, input, phantomId)) + + def bind(agentId: AgentId): Either[GolemReflectError, ReflectedAgentClient] = + for { + parts <- agentId.parts + _ <- Either.cond( + parts.typeName == name, + (), + GolemReflectError.Identity(s"Agent type '$name' cannot bind '${parts.typeName}'") + ) + _ <- Either.cond( + mode == AgentMode.Durable, + (), + GolemReflectError.Identity(s"Cannot bind an existing identity to ephemeral agent type '$name'") + ) + client <- client.createValue(parts.constructorValue, parts.phantomId) + } yield client +} + +object Reflection { + def getAllAgentTypes(): Either[GolemReflectError, List[AgentType]] = + try sequence(AgentHostApi.getAllAgentTypes().map(decodeAgentType)) + catch { case NonFatal(error) => Left(GolemReflectError.Discovery(error.getMessage)) } + + def getAgentType(name: String): Either[GolemReflectError, Option[AgentType]] = + try + AgentHostApi + .registeredAgentType(name) + .map(decodeAgentType) + .fold[Either[GolemReflectError, Option[AgentType]]](Right(None))(_.map(Some(_))) + catch { case NonFatal(error) => Left(GolemReflectError.Discovery(error.getMessage)) } + + private def decodeAgentType(registered: AgentHostApi.RegisteredAgentType): Either[GolemReflectError, AgentType] = + try { + val raw = registered.agentType + val graph = raw.schema + val methods = raw.methods.toList.map { method => + AgentMethod( + method.name, + method.description, + method.promptHint.toOption, + inputRef(graph, method.inputSchema), + outputRef(graph, method.outputSchema) + ) + } + val mode = raw.mode match { + case "durable" => AgentMode.Durable + case "ephemeral" => AgentMode.Ephemeral + case other => throw new IllegalArgumentException(s"unknown agent mode '$other'") + } + Right( + new AgentType( + raw.typeName, + raw.description, + raw.sourceLanguage, + mode, + ComponentId.fromJs(registered.implementedBy), + inputRef(graph, raw.constructor.inputSchema), + methods + ) + ) + } catch { case NonFatal(error) => Left(GolemReflectError.SchemaDecode(error.getMessage)) } + + private def inputRef(graph: JsSchemaGraph, input: JsInputSchema): SchemaRef = { + if (input.tag != "parameters") throw new IllegalArgumentException(s"unknown input schema '${input.tag}'") + val decoded = SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph)) + val entries = input.asInstanceOf[js.Dynamic].selectDynamic("val").asInstanceOf[js.Array[JsNamedField]].toList + val fields = entries.collect { + case entry if entry.source.tag == "user-supplied" => + val root = SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph).copy(root = entry.schema)).root + NamedFieldType(entry.name, root, SchemaWireInterop.metadataFromJs(entry.metadata)) + } + SchemaRef(SchemaGraph(decoded.defs, SchemaType(RecordType(fields)))) + } + + private def outputRef(graph: JsSchemaGraph, output: JsOutputSchema): Option[SchemaRef] = + output.tag match { + case "unit" => None + case "single" => + val root = output.asInstanceOf[js.Dynamic].selectDynamic("val").asInstanceOf[Int] + Some(SchemaRef(SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph).copy(root = root)))) + case other => throw new IllegalArgumentException(s"unknown output schema '$other'") + } +} + +final case class ReflectedPhantomClient(agentId: AgentId, phantomId: Uuid, client: ReflectedAgentClient) + +final class ReflectedAgentClientFactory private[reflection] (agentType: AgentType) { + def get(input: Json): Either[GolemReflectError, ReflectedAgentClient] = + requireDurable("get").flatMap(_ => pack(input)).flatMap(createValue(_, None)) + + def getValue(input: SchemaValue): Either[GolemReflectError, ReflectedAgentClient] = + requireDurable("getValue").flatMap(_ => createValue(input, None)) + + def getPhantom(input: Json, phantomId: Uuid): Either[GolemReflectError, ReflectedAgentClient] = + pack(input).flatMap(createValue(_, Some(phantomId))) + + def getPhantomValue(input: SchemaValue, phantomId: Uuid): Either[GolemReflectError, ReflectedAgentClient] = + createValue(input, Some(phantomId)) + + def newPhantom(input: Json): Either[GolemReflectError, Either[ReflectedAgentClient, ReflectedPhantomClient]] = + pack(input).flatMap(newPhantomValue) + + def newPhantomValue( + input: SchemaValue + ): Either[GolemReflectError, Either[ReflectedAgentClient, ReflectedPhantomClient]] = + if (agentType.mode == AgentMode.Ephemeral) createValue(input, None).map(Left(_)) + else { + val phantom = Uuid.random() + for { + id <- agentType.agentIdValue(input, Some(phantom)) + client <- createValue(input, Some(phantom)) + } yield Right(ReflectedPhantomClient(id, phantom, client)) + } + + private[reflection] def createValue( + input: SchemaValue, + phantomId: Option[Uuid] + ): Either[GolemReflectError, ReflectedAgentClient] = + validate(agentType.constructorInput, input) + .flatMap(_ => Transport.create(agentType.implementedBy, agentType.name, input, phantomId)) + .map(new ReflectedAgentClient(agentType, _)) + + private def pack(input: Json): Either[GolemReflectError, SchemaValue] = + agentType.constructorInput.packJson(input).left.map(error => GolemReflectError.Validation(error.message)) + + private def requireDurable(operation: String): Either[GolemReflectError, Unit] = + Either.cond( + agentType.mode == AgentMode.Durable, + (), + GolemReflectError.Identity(s"$operation is not available for ephemeral agent types") + ) +} + +final class ReflectedAgentClient private[reflection] (agentType: AgentType, transport: Transport) { + def method(name: String): Either[GolemReflectError, ReflectedAgentMethod] = + agentType + .method(name) + .toRight(GolemReflectError.Discovery(s"Agent type '${agentType.name}' has no method '$name'")) + .map(new ReflectedAgentMethod(_, transport)) +} + +final class ReflectedAgentMethod private[reflection] (val definition: AgentMethod, transport: Transport) { + def invoke(input: Json): Future[Either[GolemReflectError, Invocation[Json]]] = invokeJson(input) + + def invokeJson(input: Json): Future[Either[GolemReflectError, Invocation[Json]]] = + definition.input.packJson(input) match { + case Left(error) => Future.successful(Left(GolemReflectError.Validation(error.message))) + case Right(value) => + invokeValue(value).map(_.flatMap { invocation => + invocation.value match { + case None => Right(Invocation(invocation.metadata, None)) + case Some(result) => + definition.output + .toRight(GolemReflectError.SchemaDecode("unit method returned a value")) + .flatMap(_.unpackJson(result).left.map(error => GolemReflectError.SchemaDecode(error.message))) + .map(json => Invocation(invocation.metadata, Some(json))) + } + }) + } + + def invokeValue(input: SchemaValue): Future[Either[GolemReflectError, Invocation[SchemaValue]]] = + validate(definition.input, input) match { + case Left(error) => Future.successful(Left(error)) + case Right(_) => transport.invokeAndAwait(definition.name, input) + } + + def triggerValue(input: SchemaValue): Either[GolemReflectError, InvocationMetadata] = + validate(definition.input, input).flatMap(_ => transport.trigger(definition.name, input)) + + def triggerJson(input: Json): Either[GolemReflectError, InvocationMetadata] = + definition.input + .packJson(input) + .left + .map(error => GolemReflectError.Validation(error.message)) + .flatMap(triggerValue) + + def scheduleValue(at: Datetime, input: SchemaValue): Either[GolemReflectError, ScheduledInvocation] = + validate(definition.input, input).flatMap(_ => transport.schedule(at, definition.name, input)) + + def scheduleJson(at: Datetime, input: Json): Either[GolemReflectError, ScheduledInvocation] = + definition.input + .packJson(input) + .left + .map(error => GolemReflectError.Validation(error.message)) + .flatMap(scheduleValue(at, _)) +} + +final case class InvocationMetadata(agentId: AgentId, idempotencyKey: String) +final case class Invocation[+A](metadata: InvocationMetadata, value: Option[A]) +final case class ScheduledInvocation(metadata: InvocationMetadata, cancellationToken: CancellationToken) + +final class DynamicAgentClient private (transport: Transport, val agentId: Option[AgentId]) { + def method(name: String): DynamicAgentMethod = new DynamicAgentMethod(name, transport) +} + +object DynamicAgentClient { + def fromAgentId(agentId: AgentId): Either[GolemReflectError, DynamicAgentClient] = + agentId.parts + .flatMap(parts => Transport.create(agentId.componentId, parts.typeName, parts.constructorValue, parts.phantomId)) + .map(new DynamicAgentClient(_, Some(agentId))) + + /** + * A raw one-shot address. Final identity is supplied by invocation metadata. + */ + def ephemeral( + componentId: ComponentId, + typeName: String, + constructor: SchemaValue + ): Either[GolemReflectError, DynamicAgentClient] = + Transport.create(componentId, typeName, constructor, None).map(new DynamicAgentClient(_, None)) +} + +final class DynamicAgentMethod private[reflection] (val name: String, transport: Transport) { + def invokeValue(input: SchemaValue): Future[Either[GolemReflectError, Invocation[SchemaValue]]] = + transport.invokeAndAwait(name, input) + def triggerValue(input: SchemaValue): Either[GolemReflectError, InvocationMetadata] = transport.trigger(name, input) + def scheduleValue(at: Datetime, input: SchemaValue): Either[GolemReflectError, ScheduledInvocation] = + transport.schedule(at, name, input) +} + +private[reflection] final class Transport private (componentId: ComponentId, raw: WasmRpcApi.WasmRpcClient) { + def invokeAndAwait(method: String, input: SchemaValue): Future[Either[GolemReflectError, Invocation[SchemaValue]]] = + encodeAsync(input).flatMap { payload => + raw.asyncInvokeAndAwaitWithMetadata(method, payload) match { + case Left(error) => Future.successful(Left(GolemReflectError.Remote(error.toString))) + case Right((metadata, pending)) => + FutureInterop + .fromPromise(pending.get()) + .map { result => + decodeOptional(result.toOption).map(value => Invocation(toMetadata(metadata), value)) + } + .recover { case NonFatal(error) => Left(GolemReflectError.Remote(error.getMessage)) } + } + }.recover { case NonFatal(error) => Left(GolemReflectError.SchemaEncode(error.getMessage)) } + + def trigger(method: String, input: SchemaValue): Either[GolemReflectError, InvocationMetadata] = + encode(input).flatMap(payload => + raw + .invokeWithMetadata(method, payload) + .left + .map(error => GolemReflectError.Remote(error.toString)) + .map(toMetadata) + ) + + def schedule(at: Datetime, method: String, input: SchemaValue): Either[GolemReflectError, ScheduledInvocation] = + encode(input).flatMap(payload => + raw + .scheduleCancelableInvocationWithMetadata(at, method, payload) + .left + .map(error => GolemReflectError.Remote(error.toString)) + .map(receipt => ScheduledInvocation(toMetadata(receipt.metadata), receipt.cancellationToken)) + ) + + private def toMetadata(value: golem.runtime.rpc.InvocationMetadata): InvocationMetadata = + InvocationMetadata(AgentId(componentId, value.agentId), value.idempotencyKey) +} + +private[reflection] object Transport { + def create( + componentId: ComponentId, + typeName: String, + constructor: SchemaValue, + phantom: Option[Uuid] + ): Either[GolemReflectError, Transport] = + encode(constructor).map { payload => + val phantomArg = phantom.fold[js.UndefOr[JsSchemaUuid]](js.undefined)(uuid => + JsSchemaUuid(js.BigInt(uuid.highBits.toString), js.BigInt(uuid.lowBits.toString)) + ) + new Transport(componentId, WasmRpcApi.newClient(typeName, payload, phantomArg, js.Array())) + } +} + +private object ReflectionInternals { + def validate(schema: SchemaRef, value: SchemaValue): Either[GolemReflectError, Unit] = + schema + .validateValue(value) + .left + .map(errors => GolemReflectError.Validation(errors.map(_.message).mkString("; "))) + .map(_ => ()) + + def encode(value: SchemaValue): Either[GolemReflectError, JsSchemaValueTree] = + try Right(SchemaWireInterop.valueTreeToJs(SchemaWire.schemaValueToWit(value))) + catch { case NonFatal(error) => Left(GolemReflectError.SchemaEncode(error.getMessage)) } + + def encodeAsync(value: SchemaValue): Future[JsSchemaValueTree] = + SchemaWireInterop.valueTreeToJsAsync(SchemaWire.schemaValueToWit(value)) + + def decodeOptional(value: Option[JsSchemaValueTree]): Either[GolemReflectError, Option[SchemaValue]] = + try Right(value.map(tree => SchemaWire.schemaValueFromWit(SchemaWireInterop.valueTreeFromJs(tree)))) + catch { case NonFatal(error) => Left(GolemReflectError.SchemaDecode(error.getMessage)) } + + def sequence[A](values: List[Either[GolemReflectError, A]]): Either[GolemReflectError, List[A]] = + values.foldRight[Either[GolemReflectError, List[A]]](Right(Nil))((entry, result) => + entry.flatMap(value => result.map(value :: _)) + ) +} diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala new file mode 100644 index 0000000000..05c47312d1 --- /dev/null +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala @@ -0,0 +1,588 @@ +/* + * Copyright 2024-2026 Golem Cloud + * + * Licensed under the Golem Source License v1.1 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://license.golem.cloud/LICENSE + */ + +package golem.reflection + +import golem.schema._ +import golem.schema.SchemaTypeBody._ +import golem.schema.SchemaValue._ +import golem.schema.validation.ValueValidation +import zio.blocks.schema.json.Json + +import scala.collection.immutable.ListMap + +final case class SchemaIssue(message: String, path: List[String] = Nil) + +final class SchemaRef private (val graph: SchemaGraph, val root: SchemaType) { + def validateValue(value: SchemaValue): Either[List[SchemaIssue], SchemaValue] = + ValueValidation + .validateValue(graph, root, value) + .left + .map(_.map(error => SchemaIssue(error.message))) + .map(_ => value) + + def validateJson(value: Json): Either[List[SchemaIssue], SchemaValue] = + packJson(value).left.map(error => List(error)).flatMap(validateValue) + + def packJson(value: Json): Either[SchemaIssue, SchemaValue] = + CanonicalJson.pack(graph, root, value) + + def unpackJson(value: SchemaValue): Either[SchemaIssue, Json] = + validateValue(value).left.map(_.head).flatMap(_ => CanonicalJson.unpack(graph, root, value)) + + def toJsonSchema(includeDraftMarker: Boolean = true): Json = + CanonicalJson.jsonSchema(graph, root, includeDraftMarker) +} + +object SchemaRef { + def apply(graph: SchemaGraph): SchemaRef = new SchemaRef(graph, graph.root) + + def apply(graph: SchemaGraph, root: SchemaType): SchemaRef = + new SchemaRef(SchemaGraph(graph.defs, root), root) +} + +private object CanonicalJson { + private val MaxSafeInteger = BigInt("9007199254740991") + + def pack(graph: SchemaGraph, schema: SchemaType, json: Json): Either[SchemaIssue, SchemaValue] = + attempt(packUnsafe(graph, resolve(graph, schema), json)) + + def unpack(graph: SchemaGraph, schema: SchemaType, value: SchemaValue): Either[SchemaIssue, Json] = + attempt(unpackUnsafe(graph, resolve(graph, schema), value)) + + def jsonSchema(graph: SchemaGraph, schema: SchemaType, includeDraftMarker: Boolean): Json = { + val root = schemaJson(graph, schema) + val definitions = Json.Object(graph.defs.toList.map { case (id, definition) => + id -> schemaJson(graph, definition.body) + }: _*) + val rootFields = fields(root).toList ++ + (if (includeDraftMarker) List("$schema" -> Json.String("https://json-schema.org/draft/2020-12/schema")) + else Nil) ++ + (if (graph.defs.isEmpty) Nil else List("$defs" -> definitions)) + Json.Object(rootFields: _*) + } + + private def attempt[A](value: => A): Either[SchemaIssue, A] = + try Right(value) + catch { case error: IllegalArgumentException => Left(SchemaIssue(error.getMessage)) } + + private def fail(message: String): Nothing = throw new IllegalArgumentException(message) + + private def schemaJson(graph: SchemaGraph, schema: SchemaType): Json = { + def typed(name: String, extra: (String, Json)*): Json = Json.Object(("type" -> Json.String(name)) +: extra: _*) + def integer(min: BigInt, max: BigInt): Json = + typed("integer", "minimum" -> number(BigDecimal(min)), "maximum" -> number(BigDecimal(max))) + schema.body match { + case RefType(id) => Json.Object("$ref" -> Json.String(s"#/$$defs/${id.replace("~", "~0").replace("/", "~1")}")) + case BoolType => typed("boolean") + case S8Type(_) => integer(-128, 127) + case S16Type(_) => integer(-32768, 32767) + case S32Type(_) => integer(Int.MinValue, Int.MaxValue) + case S64Type(_) => integer(BigInt(Long.MinValue), BigInt(Long.MaxValue)) + case U8Type(_) => integer(0, 255) + case U16Type(_) => integer(0, 65535) + case U32Type(_) => integer(0, BigInt("4294967295")) + case U64Type(_) => integer(0, (BigInt(1) << 64) - 1) + case F32Type(_) | F64Type(_) => typed("number") + case CharType => typed("string", "minLength" -> number(1), "maxLength" -> number(1)) + case StringType => typed("string") + case RecordType(recordFields) => + Json.Object( + "type" -> Json.String("object"), + "properties" -> Json.Object(recordFields.map(field => field.name -> schemaJson(graph, field.body)): _*), + "required" -> Json.Array(recordFields.collect { + case field if resolve(graph, field.body).body match { case OptionType(_) => false; case _ => true } => + Json.String(field.name) + }: _*), + "additionalProperties" -> Json.Boolean(false) + ) + case VariantType(cases) => + Json.Object("oneOf" -> Json.Array(cases.map { entry => + entry.payload match { + case None => Json.Object("const" -> Json.String(entry.name)) + case Some(payload) => + Json.Object( + "type" -> Json.String("object"), + "properties" -> Json.Object(entry.name -> schemaJson(graph, payload)), + "required" -> Json.Array(Json.String(entry.name)), + "additionalProperties" -> Json.Boolean(false) + ) + } + }: _*)) + case EnumType(cases) => typed("string", "enum" -> Json.Array(cases.map(Json.String): _*)) + case FlagsType(names) => + typed( + "array", + "items" -> typed("string", "enum" -> Json.Array(names.map(Json.String): _*)), + "uniqueItems" -> Json.Boolean(true) + ) + case TupleType(elements) => + typed( + "array", + "prefixItems" -> Json.Array(elements.map(schemaJson(graph, _)): _*), + "items" -> Json.Boolean(false), + "minItems" -> number(elements.size), + "maxItems" -> number(elements.size) + ) + case ListType(element) => typed("array", "items" -> schemaJson(graph, element)) + case FixedListType(element, length) => + typed( + "array", + "items" -> schemaJson(graph, element), + "minItems" -> number(length), + "maxItems" -> number(length) + ) + case MapType(key, value) => + typed( + "array", + "items" -> typed( + "array", + "prefixItems" -> Json.Array(schemaJson(graph, key), schemaJson(graph, value)), + "items" -> Json.Boolean(false), + "minItems" -> number(2), + "maxItems" -> number(2) + ) + ) + case OptionType(element) => Json.Object("oneOf" -> Json.Array(typed("null"), schemaJson(graph, element))) + case ResultType(ok, err) => + def side(name: String, payload: Option[SchemaType]): Json = Json.Object( + "type" -> Json.String("object"), + "properties" -> Json.Object(name -> payload.fold[Json](typed("null"))(schemaJson(graph, _))), + "required" -> Json.Array(Json.String(name)), + "additionalProperties" -> Json.Boolean(false) + ) + Json.Object("oneOf" -> Json.Array(side("ok", ok), side("err", err))) + case TextType(_) => + typed( + "object", + "properties" -> Json.Object("text" -> typed("string"), "language" -> typed("string")), + "required" -> Json.Array(Json.String("text")), + "additionalProperties" -> Json.Boolean(false) + ) + case BinaryType(_) => + typed( + "object", + "properties" -> Json.Object( + "bytes" -> typed("string", "contentEncoding" -> Json.String("base64url")), + "mimeType" -> typed("string") + ), + "required" -> Json.Array(Json.String("bytes")), + "additionalProperties" -> Json.Boolean(false) + ) + case PathType(_) => typed("string", "format" -> Json.String("file-path")) + case UrlType(_) => typed("string", "format" -> Json.String("uri")) + case DatetimeType => typed("string", "format" -> Json.String("date-time")) + case DurationType => typed("string", "format" -> Json.String("duration")) + case QuantityType(_) => + typed( + "object", + "properties" -> Json + .Object("mantissa" -> typed("integer"), "scale" -> typed("integer"), "unit" -> typed("string")), + "required" -> Json.Array(Json.String("mantissa"), Json.String("scale"), Json.String("unit")), + "additionalProperties" -> Json.Boolean(false) + ) + case UnionType(branches) => + Json.Object("oneOf" -> Json.Array(branches.map(branch => schemaJson(graph, branch.body)): _*)) + case SecretType(_) => Json.Object("x-golem-capability" -> Json.String("secret")) + case QuotaTokenType(_) => Json.Object("x-golem-capability" -> Json.String("quota-token")) + case PermissionCardType(_) => Json.Object("x-golem-capability" -> Json.String("permission-card")) + case FutureType(_) => Json.Object("x-golem-unsupported" -> Json.String("future")) + case StreamType(_) => Json.Object("x-golem-unsupported" -> Json.String("stream")) + } + } + + private def resolve(graph: SchemaGraph, schema: SchemaType, seen: Set[String] = Set.empty): SchemaType = + schema.body match { + case RefType(id) if seen(id) => fail(s"reference cycle through '$id'") + case RefType(id) => + graph.defs.get(id) match { + case Some(definition) => resolve(graph, definition.body, seen + id) + case None => fail(s"dangling reference '$id'") + } + case _ => schema + } + + private def fields(json: Json): ListMap[String, Json] = json match { + case Json.Object(values) => ListMap(values.toList: _*) + case _ => fail("expected a JSON object") + } + + private def array(json: Json): List[Json] = json match { + case Json.Array(values) => values.toList + case _ => fail("expected a JSON array") + } + + private def string(json: Json): String = json match { + case value: Json.String => value.value + case _ => fail("expected a JSON string") + } + + private def bool(json: Json): Boolean = json match { + case value: Json.Boolean => value.value + case _ => fail("expected a JSON boolean") + } + + private def decimal(json: Json): BigDecimal = json match { + case Json.Number(value) => value + case _ => fail("expected a JSON number") + } + + private def integral(json: Json, min: BigInt, max: BigInt): BigInt = { + val value = decimal(json) + if (!value.isWhole) fail("expected an integer") + val result = value.toBigInt + if (result < min || result > max) fail(s"integer is outside [$min, $max]") + result + } + + private def safeLong(json: Json, unsigned: Boolean): Long = { + val min = if (unsigned) BigInt(0) else -MaxSafeInteger + val value = integral(json, min, MaxSafeInteger) + value.toLong + } + + private def packUnsafe(graph: SchemaGraph, schema: SchemaType, json: Json): SchemaValue = + schema.body match { + case BoolType => BoolValue(bool(json)) + case S8Type(_) => S8Value(integral(json, -128, 127).toByte) + case S16Type(_) => S16Value(integral(json, -32768, 32767).toShort) + case S32Type(_) => S32Value(integral(json, Int.MinValue, Int.MaxValue).toInt) + case S64Type(_) => S64Value(safeLong(json, unsigned = false)) + case U8Type(_) => U8Value(integral(json, 0, 255).toInt) + case U16Type(_) => U16Value(integral(json, 0, 65535).toInt) + case U32Type(_) => U32Value(integral(json, 0, BigInt("4294967295")).toLong) + case U64Type(_) => U64Value(safeLong(json, unsigned = true)) + case F32Type(_) => F32Value(decimal(json).toFloat) + case F64Type(_) => F64Value(decimal(json).toDouble) + case CharType => + val text = string(json) + if (text.codePointCount(0, text.length) != 1) fail("expected one Unicode scalar") + CharValue(text.codePointAt(0)) + case StringType => StringValue(string(json)) + case RecordType(expected) => + val jsonFields = fields(json) + jsonFields.keys.find(name => !expected.exists(_.name == name)).foreach(name => fail(s"unknown field '$name'")) + RecordValue( + expected.map(field => + packUnsafe( + graph, + resolve(graph, field.body), + jsonFields.getOrElse(field.name, fail(s"missing field '${field.name}'")) + ) + ) + ) + case VariantType(cases) => + json match { + case value: Json.String => + val index = cases.indexWhere(entry => entry.name == value.value && entry.payload.isEmpty) + if (index < 0) fail(s"unknown payload-free variant case '${value.value}'") + VariantValue(index, None) + case _ => + val jsonFields = fields(json) + if (jsonFields.size != 1) fail("expected a single-key variant object") + val (name, payload) = jsonFields.head + val index = cases.indexWhere(_.name == name) + if (index < 0 || cases(index).payload.isEmpty) fail(s"unknown payload variant case '$name'") + VariantValue(index, Some(packUnsafe(graph, resolve(graph, cases(index).payload.get), payload))) + } + case EnumType(cases) => + val name = string(json) + val index = cases.indexOf(name) + if (index < 0) fail(s"unknown enum case '$name'") + EnumValue(index) + case FlagsType(names) => + val selected = array(json).map(string) + selected.find(!names.contains(_)).foreach(name => fail(s"unknown flag '$name'")) + if (selected.distinct.size != selected.size) fail("duplicate flag") + FlagsValue(names.map(selected.contains)) + case TupleType(elements) => + val values = array(json) + if (values.size != elements.size) fail(s"expected ${elements.size} tuple elements") + TupleValue(elements.zip(values).map { case (entry, value) => packUnsafe(graph, resolve(graph, entry), value) }) + case ListType(element) => + ListValue(array(json).map(value => packUnsafe(graph, resolve(graph, element), value))) + case FixedListType(element, length) => + val values = array(json) + if (values.size != length) fail(s"expected $length elements") + FixedListValue(values.map(value => packUnsafe(graph, resolve(graph, element), value))) + case MapType(key, value) => + MapValue(array(json).map { entry => + val pair = array(entry) + if (pair.size != 2) fail("expected a two-element map entry") + SchemaMapEntry( + packUnsafe(graph, resolve(graph, key), pair.head), + packUnsafe(graph, resolve(graph, value), pair(1)) + ) + }) + case OptionType(element) => + json match { + case Json.Null => OptionValue(None) + case other => OptionValue(Some(packUnsafe(graph, resolve(graph, element), other))) + } + case ResultType(ok, err) => + val jsonFields = fields(json) + if (jsonFields.size != 1 || !Set("ok", "err")(jsonFields.head._1)) fail("expected {'ok': ...} or {'err': ...}") + val (side, payload) = jsonFields.head + val expected = if (side == "ok") ok else err + val packed = expected match { + case None if payload == Json.Null => None + case None => fail("expected null unit payload") + case Some(value) => Some(packUnsafe(graph, resolve(graph, value), payload)) + } + ResultValue(if (side == "ok") SchemaResult.Ok(packed) else SchemaResult.Err(packed)) + case TextType(_) => + val jsonFields = fields(json) + TextValue( + string(jsonFields.getOrElse("text", fail("missing field 'text'"))), + jsonFields.get("language").map(string) + ) + case BinaryType(_) => + val jsonFields = fields(json) + BinaryValue( + decodeBase64Url(string(jsonFields.getOrElse("bytes", fail("missing field 'bytes'")))), + jsonFields.get("mimeType").map(string) + ) + case PathType(_) => PathValue(string(json)) + case UrlType(_) => UrlValue(string(json)) + case DatetimeType => + val instant = java.time.Instant.parse(string(json)) + DatetimeValue(Datetime(instant.getEpochSecond, instant.getNano)) + case DurationType => DurationValue(decodeDuration(string(json))) + case QuantityType(_) => + val jsonFields = fields(json) + QuantityValueNode( + QuantityValue( + safeLong(jsonFields("mantissa"), unsigned = false), + integral(jsonFields("scale"), Int.MinValue, Int.MaxValue).toInt, + string(jsonFields("unit")) + ) + ) + case UnionType(branches) => + val matching = branches.filter(branch => discriminatorMatches(branch.discriminator, json)) + if (matching.size != 1) fail(s"expected exactly one matching union branch, found ${matching.size}") + UnionValue(matching.head.tag, packUnsafe(graph, resolve(graph, matching.head.body), json)) + case SecretType(_) | QuotaTokenType(_) | PermissionCardType(_) => + fail("capability values cannot be constructed from JSON") + case FutureType(_) | StreamType(_) => fail("future and stream values have no JSON representation") + case RefType(_) => fail("unresolved schema reference") + } + + private def number(value: BigDecimal): Json = Json.Number(value) + + private def unpackUnsafe(graph: SchemaGraph, schema: SchemaType, value: SchemaValue): Json = + (schema.body, value) match { + case (BoolType, BoolValue(x)) => Json.Boolean(x) + case (S8Type(_), S8Value(x)) => number(BigDecimal(x)) + case (S16Type(_), S16Value(x)) => number(BigDecimal(x)) + case (S32Type(_), S32Value(x)) => number(BigDecimal(x)) + case (S64Type(_), S64Value(x)) if BigInt(x).abs <= MaxSafeInteger => number(BigDecimal(x)) + case (U8Type(_), U8Value(x)) => number(BigDecimal(x)) + case (U16Type(_), U16Value(x)) => number(BigDecimal(x)) + case (U32Type(_), U32Value(x)) => number(BigDecimal(x)) + case (U64Type(_), U64Value(x)) if x >= 0 && BigInt(x) <= MaxSafeInteger => number(BigDecimal(x)) + case (F32Type(_), F32Value(x)) => number(BigDecimal.decimal(x)) + case (F64Type(_), F64Value(x)) => number(BigDecimal(x)) + case (CharType, CharValue(x)) => Json.String(new String(Character.toChars(x))) + case (StringType, StringValue(x)) => Json.String(x) + case (RecordType(expected), RecordValue(values)) if expected.size == values.size => + Json.Object(expected.zip(values).map { case (field, entry) => + field.name -> unpackUnsafe(graph, resolve(graph, field.body), entry) + }: _*) + case (VariantType(cases), VariantValue(index, payload)) if cases.isDefinedAt(index) => + val entry = cases(index) + payload match { + case None => Json.String(entry.name) + case Some(inner) => + Json.Object( + entry.name -> unpackUnsafe( + graph, + resolve(graph, entry.payload.getOrElse(fail("unexpected variant payload"))), + inner + ) + ) + } + case (EnumType(cases), EnumValue(index)) if cases.isDefinedAt(index) => Json.String(cases(index)) + case (FlagsType(names), FlagsValue(bits)) if names.size == bits.size => + Json.Array(names.zip(bits).collect { case (name, true) => Json.String(name) }: _*) + case (TupleType(types), TupleValue(values)) if types.size == values.size => + Json.Array( + types.zip(values).map { case (entry, inner) => unpackUnsafe(graph, resolve(graph, entry), inner) }: _* + ) + case (ListType(element), ListValue(values)) => + Json.Array(values.map(unpackUnsafe(graph, resolve(graph, element), _)): _*) + case (FixedListType(element, length), FixedListValue(values)) if values.size == length => + Json.Array(values.map(unpackUnsafe(graph, resolve(graph, element), _)): _*) + case (MapType(key, entry), MapValue(values)) => + Json.Array( + values.map(value => + Json.Array( + unpackUnsafe(graph, resolve(graph, key), value.key), + unpackUnsafe(graph, resolve(graph, entry), value.value) + ) + ): _* + ) + case (OptionType(_), OptionValue(None)) => Json.Null + case (OptionType(element), OptionValue(Some(inner))) => unpackUnsafe(graph, resolve(graph, element), inner) + case (ResultType(ok, _), ResultValue(SchemaResult.Ok(inner))) => + val rendered = inner match { + case None => Json.Null + case Some(value) => unpackUnsafe(graph, resolve(graph, ok.getOrElse(fail("unexpected ok payload"))), value) + } + Json.Object("ok" -> rendered) + case (ResultType(_, err), ResultValue(SchemaResult.Err(inner))) => + val rendered = inner match { + case None => Json.Null + case Some(value) => unpackUnsafe(graph, resolve(graph, err.getOrElse(fail("unexpected err payload"))), value) + } + Json.Object("err" -> rendered) + case (TextType(_), TextValue(text, language)) => + Json.Object((List("text" -> Json.String(text)) ++ language.map(value => "language" -> Json.String(value))): _*) + case (BinaryType(_), BinaryValue(bytes, mimeType)) => + Json.Object( + (List("bytes" -> Json.String(encodeBase64Url(bytes))) ++ mimeType.map(value => + "mimeType" -> Json.String(value) + )): _* + ) + case (PathType(_), PathValue(x)) => Json.String(x) + case (UrlType(_), UrlValue(x)) => Json.String(x) + case (DatetimeType, DatetimeValue(x)) => + Json.String(java.time.Instant.ofEpochSecond(x.seconds, x.nanoseconds.toLong).toString) + case (DurationType, DurationValue(x)) => Json.String(encodeDuration(x)) + case (QuantityType(_), QuantityValueNode(x)) => + Json.Object( + "mantissa" -> number(BigDecimal(x.mantissa)), + "scale" -> number(BigDecimal(x.scale)), + "unit" -> Json.String(x.unit) + ) + case (UnionType(branches), UnionValue(tag, body)) => + val branch = branches.find(_.tag == tag).getOrElse(fail(s"unknown union tag '$tag'")) + unpackUnsafe(graph, resolve(graph, branch.body), body) + case (SecretType(_), _) | (QuotaTokenType(_), _) | (PermissionCardType(_), _) => + fail("capability values cannot be exposed as JSON") + case (FutureType(_), _) | (StreamType(_), _) => fail("future and stream values have no JSON representation") + case _ => fail(s"schema value does not match ${schema.body}") + } + + private def discriminatorMatches(rule: DiscriminatorRule, json: Json): Boolean = rule match { + case DiscriminatorRule.Prefix(value) => + json match { + case text: Json.String => text.value.startsWith(value) + case _ => false + } + case DiscriminatorRule.Suffix(value) => + json match { + case text: Json.String => text.value.endsWith(value) + case _ => false + } + case DiscriminatorRule.Contains(value) => + json match { + case text: Json.String => text.value.contains(value) + case _ => false + } + case DiscriminatorRule.Regex(value) => + json match { + case text: Json.String => value.r.findFirstIn(text.value).nonEmpty + case _ => false + } + case DiscriminatorRule.FieldEquals(field) => + json match { + case Json.Object(values) => + values.toList.find(_._1 == field.fieldName).exists { case (_, value) => + field.literal.forall(expected => value == Json.String(expected)) + } + case _ => false + } + case DiscriminatorRule.FieldAbsent(name) => + json match { + case Json.Object(values) => !values.toList.exists(_._1 == name) + case _ => false + } + } + + private val Base64Alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_" + + private def encodeBase64Url(bytes: Vector[Byte]): String = { + val result = new StringBuilder + var index = 0 + while (index < bytes.length) { + val first = bytes(index) & 0xff + val second = if (index + 1 < bytes.length) bytes(index + 1) & 0xff else 0 + val third = if (index + 2 < bytes.length) bytes(index + 2) & 0xff else 0 + result += Base64Alphabet.charAt(first >> 2) + result += Base64Alphabet.charAt(((first & 3) << 4) | (second >> 4)) + if (index + 1 < bytes.length) result += Base64Alphabet.charAt(((second & 15) << 2) | (third >> 6)) + if (index + 2 < bytes.length) result += Base64Alphabet.charAt(third & 63) + index += 3 + } + result.result() + } + + private def decodeBase64Url(value: String): Vector[Byte] = { + if (!value.forall(Base64Alphabet.contains(_)) || value.length % 4 == 1) + fail("invalid base64url without padding") + val result = Vector.newBuilder[Byte] + var index = 0 + while (index < value.length) { + val a = Base64Alphabet.indexOf(value(index)) + val b = Base64Alphabet.indexOf(value(index + 1)) + val c = if (index + 2 < value.length) Base64Alphabet.indexOf(value(index + 2)) else 0 + val d = if (index + 3 < value.length) Base64Alphabet.indexOf(value(index + 3)) else 0 + result += ((a << 2) | (b >> 4)).toByte + if (index + 2 < value.length) result += (((b & 15) << 4) | (c >> 2)).toByte + if (index + 3 < value.length) result += (((c & 3) << 6) | d).toByte + index += 4 + } + result.result() + } + + private val DurationPattern = + "^(-)?P(?:(\\d+)D)?(?:T(?:(\\d+)H)?(?:(\\d+)M)?(?:(\\d+)(?:\\.(\\d{1,9}))?S)?)?$".r + + private def decodeDuration(value: String): Long = value match { + case DurationPattern(sign, days, hours, minutes, seconds, fraction) => + if (List(days, hours, minutes, seconds).forall(_ == null)) fail("expected an ISO 8601 duration") + def amount(raw: String): BigInt = if (raw == null) BigInt(0) else BigInt(raw) + val fractional = Option(fraction).fold(BigInt(0))(raw => BigInt(raw.padTo(9, '0').mkString)) + val nanos = amount(days) * 86400000000000L + amount(hours) * 3600000000000L + + amount(minutes) * 60000000000L + amount(seconds) * 1000000000L + fractional + val signed = if (sign == null) nanos else -nanos + if (!signed.isValidLong) fail("duration nanoseconds out of i64 range") + signed.toLong + case _ => fail("expected an ISO 8601 duration") + } + + private def encodeDuration(nanoseconds: Long): String = + if (nanoseconds == 0) "PT0S" + else { + val negative = nanoseconds < 0 + var remaining = BigInt(nanoseconds).abs + val days = remaining / 86400000000000L + remaining %= 86400000000000L + val hours = remaining / 3600000000000L + remaining %= 3600000000000L + val minutes = remaining / 60000000000L + remaining %= 60000000000L + val seconds = remaining / 1000000000L + val nanos = remaining % 1000000000L + val result = new StringBuilder(if (negative) "-P" else "P") + if (days != 0) result.append(days).append('D') + if (hours != 0 || minutes != 0 || seconds != 0 || nanos != 0) { + result.append('T') + if (hours != 0) result.append(hours).append('H') + if (minutes != 0) result.append(minutes).append('M') + if (seconds != 0 || nanos != 0) { + result.append(seconds) + if (nanos != 0) result.append('.').append(f"${nanos.toLong}%09d".reverse.dropWhile(_ == '0').reverse) + result.append('S') + } + } + result.result() + } +} diff --git a/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/WasmRpcApi.scala b/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/WasmRpcApi.scala index dc00a54c41..ade7d2a151 100644 --- a/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/WasmRpcApi.scala +++ b/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/WasmRpcApi.scala @@ -237,7 +237,7 @@ private[golem] object WasmRpcApi { @js.native @JSImport("golem:agent/host@2.0.0", "FutureInvokeResult") - private[rpc] class RawFutureInvokeResult extends js.Object { + private[golem] class RawFutureInvokeResult extends js.Object { def get(): js.Promise[js.UndefOr[JsSchemaValueTree]] = js.native def cancel(): Unit = js.native } diff --git a/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala b/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala new file mode 100644 index 0000000000..3293b06cc4 --- /dev/null +++ b/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala @@ -0,0 +1,102 @@ +/* + * Copyright 2024-2026 Golem Cloud + * + * Licensed under the Golem Source License v1.1 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://license.golem.cloud/LICENSE + */ + +package golem.reflection + +import golem.schema._ +import golem.schema.SchemaTypeBody._ +import golem.schema.SchemaValue._ +import zio.test._ +import zio.blocks.schema.json.Json + +import scala.collection.immutable.ListMap + +object SchemaRefSpec extends ZIOSpecDefault { + private val graph = SchemaGraph( + ListMap.empty, + SchemaType( + RecordType( + List( + NamedFieldType("name", SchemaType(StringType)), + NamedFieldType("count", SchemaType(U32Type())), + NamedFieldType("enabled", SchemaType(BoolType)), + NamedFieldType("labels", SchemaType(ListType(SchemaType(StringType)))) + ) + ) + ) + ) + + def spec = suite("SchemaRef")( + test("packs and unpacks canonical record JSON") { + val ref = SchemaRef(graph) + val json = Json.Object( + "name" -> Json.String("worker"), + "count" -> Json.Number(BigDecimal(42)), + "enabled" -> Json.Boolean(true), + "labels" -> Json.Array(Json.String("a"), Json.String("b")) + ) + val expected = RecordValue( + List(StringValue("worker"), U32Value(42), BoolValue(true), ListValue(List(StringValue("a"), StringValue("b")))) + ) + assertTrue(ref.packJson(json) == Right(expected), ref.unpackJson(expected) == Right(json)) + }, + test("rejects unknown fields and invalid direct values") { + val ref = SchemaRef(graph) + val invalidJson = Json.Object( + "name" -> Json.String("worker"), + "count" -> Json.Number(BigDecimal(1)), + "enabled" -> Json.Boolean(true), + "labels" -> Json.Array(), + "extra" -> Json.String("no") + ) + assertTrue( + ref.packJson(invalidJson).isLeft, + ref.validateValue(RecordValue(List(StringValue("too-short")))).isLeft + ) + }, + test("round-trips rich canonical JSON values") { + val rich = SchemaRef( + SchemaGraph( + ListMap.empty, + SchemaType( + TupleType( + List( + SchemaType(BinaryType(BinaryRestrictions.empty)), + SchemaType(DatetimeType), + SchemaType(DurationType) + ) + ) + ) + ) + ) + val value = TupleValue( + List( + BinaryValue(Vector(0, 1, 2, -1), Some("application/octet-stream")), + DatetimeValue(Datetime(1704067200L, 123000000)), + DurationValue(3723000000004L) + ) + ) + assertTrue(rich.unpackJson(value).flatMap(rich.packJson) == Right(value)) + }, + test("does not expose capabilities as JSON") { + val ref = SchemaRef( + SchemaGraph(ListMap.empty, SchemaType(PermissionCardType(PermissionCardSpec(polymorphic = false)))) + ) + assertTrue(ref.packJson(Json.Null).isLeft) + }, + test("renders canonical JSON Schema") { + val rendered = SchemaRef(graph).toJsonSchema() + assertTrue( + rendered.get("$schema").one == Right(Json.String("https://json-schema.org/draft/2020-12/schema")), + rendered.get("type").one == Right(Json.String("object")) + ) + } + ) +} From 79a597566e6413505743e53fdb4e747a6d8f6cbe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 5 Sep 2026 13:10:22 +0200 Subject: [PATCH 2/4] Align Scala reflection with the common contract --- docs/src/content/next/how-to-guides/scala.mdx | 2 +- .../content/next/how-to-guides/scala/_meta.js | 2 +- ...a.mdx => golem-agent-reflection-scala.mdx} | 20 ++++-- .../SKILL.md | 22 +++++-- .../scenarios/rpc-5-runtime-reflection.yaml | 2 +- .../golem/reflection/CallerCodecClient.scala | 50 ++++++++++++--- .../scala/golem/reflection/Reflection.scala | 63 ++++++++++++++----- .../scala/golem/reflection/SchemaRef.scala | 2 + .../golem/reflection/SchemaRefSpec.scala | 27 ++++++++ 9 files changed, 153 insertions(+), 37 deletions(-) rename docs/src/content/next/how-to-guides/scala/{golem-call-agent-reflectively-scala.mdx => golem-agent-reflection-scala.mdx} (75%) rename golem-skills/skills/scala/{golem-call-agent-reflectively-scala => golem-agent-reflection-scala}/SKILL.md (76%) diff --git a/docs/src/content/next/how-to-guides/scala.mdx b/docs/src/content/next/how-to-guides/scala.mdx index 25ff65ec16..8deefc17ec 100644 --- a/docs/src/content/next/how-to-guides/scala.mdx +++ b/docs/src/content/next/how-to-guides/scala.mdx @@ -15,7 +15,7 @@ Guides specific to developing Golem agents in Scala. - + diff --git a/docs/src/content/next/how-to-guides/scala/_meta.js b/docs/src/content/next/how-to-guides/scala/_meta.js index e5ce891992..f8875c0ba7 100644 --- a/docs/src/content/next/how-to-guides/scala/_meta.js +++ b/docs/src/content/next/how-to-guides/scala/_meta.js @@ -9,7 +9,7 @@ export default { "golem-annotate-agent-scala": "Annotating Agent Methods (Scala)", "golem-atomic-block-scala": "Atomic Blocks and Durability Controls (Scala)", "golem-call-from-external-scala": "Calling Agents from External Applications (Scala)", - "golem-call-agent-reflectively-scala": "Calling Agents with Runtime Reflection (Scala)", + "golem-agent-reflection-scala": "Calling Agents with Runtime Reflection (Scala)", "golem-call-another-agent-scala": "Calling Another Agent (Scala)", "golem-configure-durability-scala": "Configuring Agent Durability (Scala)", "golem-add-cors-scala": "Configuring CORS for Scala HTTP Endpoints", diff --git a/docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx b/docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx similarity index 75% rename from docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx rename to docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx index 03a1e36954..24d8a6d074 100644 --- a/docs/src/content/next/how-to-guides/scala/golem-call-agent-reflectively-scala.mdx +++ b/docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx @@ -18,9 +18,12 @@ val method = target.flatMap( ) ``` -An `AgentType` exposes its component ID, lifecycle mode, constructor +Agent type names are unique in an environment. An `AgentType` exposes its +current component ID, lifecycle mode, constructor `SchemaRef`, and method `SchemaRef`s. `SchemaRef` validates `SchemaValue`, packs and unpacks canonical `zio.blocks.schema.json.Json`, and renders JSON Schema. +`getAgentType` returns `Right(None)` for a missing type and reserves `Left` for +discovery or decoding failures. ## Use the three reflected/value invocation paths @@ -54,19 +57,22 @@ record fields must be packed in declaration order; the runtime authoritatively accepts or rejects the attempt. All reflected and direct methods support awaited, trigger, and scheduled calls -through `invokeValue`, `triggerValue`, and `scheduleValue`. +through `invokeValue`, `triggerValue`, and `scheduleValue`. Reflected live +streams are supported by awaited value calls. Trigger and scheduled reflected +or caller-codec calls reject methods whose input or output schema contains a +stream. ## Define a caller-codec typed contract -`AgentClientDefinition` is discovery-free. `InputRecordCodec` and -`OutputCodec` are the caller's schema authority: +`AgentClientDefinition` does not discover remote schemas. `InputRecordCodec` +and `OutputCodec` are the caller's schema authority; the environment-unique +type name is used only to resolve current implementation identity metadata: ```scala import golem.reflection._ import golem.runtime.{InputRecordCodec, OutputCodec} val contract = AgentClientDefinition( - componentId, "CounterAgent", InputRecordCodec.single[String]("name") ) @@ -80,6 +86,10 @@ val counter = contract.client.get("main") val result = counter.map(_.method(add).invoke(5)) ``` +Use `agentId.client(contract)` to bind the same caller-owned codecs to an +existing durable identity. Both creation and binding fail when the named type +is not registered in the current environment. + ## Lifecycle attempts - Use a supplied `AgentId` directly or inspect it with `parts`. diff --git a/golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md b/golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md similarity index 76% rename from golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md rename to golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md index 35f3e1fab6..0ee1655306 100644 --- a/golem-skills/skills/scala/golem-call-agent-reflectively-scala/SKILL.md +++ b/golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md @@ -1,5 +1,5 @@ --- -name: golem-call-agent-reflectively-scala +name: golem-agent-reflection-scala description: "Discovering and calling Golem agents through runtime reflection in Scala. Use when agent types or methods are selected dynamically, schemas must be inspected at runtime, caller-owned codecs are needed, or SchemaValue calls must avoid discovery." --- @@ -23,9 +23,12 @@ val method = target.flatMap( ) ``` -An `AgentType` exposes its component ID, lifecycle mode, constructor +Agent type names are unique in an environment. An `AgentType` exposes its +current component ID, lifecycle mode, constructor `SchemaRef`, and method `SchemaRef`s. `SchemaRef` validates `SchemaValue`, packs and unpacks canonical `zio.blocks.schema.json.Json`, and renders JSON Schema. +`getAgentType` returns `Right(None)` for a missing type and reserves `Left` for +discovery or decoding failures. ## Use the three reflected/value invocation paths @@ -59,19 +62,22 @@ record fields must be packed in declaration order; the runtime authoritatively accepts or rejects the attempt. All reflected and direct methods support awaited, trigger, and scheduled calls -through `invokeValue`, `triggerValue`, and `scheduleValue`. +through `invokeValue`, `triggerValue`, and `scheduleValue`. Reflected live +streams are supported by awaited value calls. Trigger and scheduled reflected +or caller-codec calls reject methods whose input or output schema contains a +stream. ## Define a caller-codec typed contract -`AgentClientDefinition` is discovery-free. `InputRecordCodec` and -`OutputCodec` are the caller's schema authority: +`AgentClientDefinition` does not discover remote schemas. `InputRecordCodec` +and `OutputCodec` are the caller's schema authority; the environment-unique +type name is used only to resolve current implementation identity metadata: ```scala import golem.reflection._ import golem.runtime.{InputRecordCodec, OutputCodec} val contract = AgentClientDefinition( - componentId, "CounterAgent", InputRecordCodec.single[String]("name") ) @@ -85,6 +91,10 @@ val counter = contract.client.get("main") val result = counter.map(_.method(add).invoke(5)) ``` +Use `agentId.client(contract)` to bind the same caller-owned codecs to an +existing durable identity. Both creation and binding fail when the named type +is not registered in the current environment. + ## Lifecycle attempts - Use a supplied `AgentId` directly or inspect it with `parts`. diff --git a/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml b/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml index 9a1abc75a3..0f065000d1 100644 --- a/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml +++ b/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml @@ -165,7 +165,7 @@ steps: `methodName`, `jsonValue`, `callerCodec`, `reflectedValue`, and `directValue`. Make sure the project builds successfully. expectedSkills: - - "golem-call-agent-reflectively-scala" + - "golem-agent-reflection-scala" verify: build: true deploy: true diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala index 1e29e67c24..477a6962a6 100644 --- a/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala @@ -20,7 +20,6 @@ import scala.util.control.NonFatal /** A discovery-free, caller-authored typed agent contract. */ final class AgentClientDefinition[Constructor] private ( - val componentId: ComponentId, val name: String, val mode: AgentMode, val constructor: InputRecordCodec[Constructor] @@ -33,16 +32,32 @@ final class AgentClientDefinition[Constructor] private ( output: OutputCodec[Output] ): CallerCodecMethod[Input, Output] = CallerCodecMethod(name, input, output) + + def bind(agentId: AgentId): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = + for { + parts <- agentId.parts + _ <- Either.cond( + parts.typeName == name, + (), + GolemReflectError.Identity(s"Agent client contract '$name' cannot bind '${parts.typeName}'") + ) + _ <- Either.cond( + mode == AgentMode.Durable, + (), + GolemReflectError.Identity(s"Cannot bind an existing identity to ephemeral agent type '$name'") + ) + componentId <- Reflection.componentIdFor(name) + transport <- Transport.create(componentId, name, parts.constructorValue, parts.phantomId) + } yield new CallerCodecAgentClient(this, transport) } object AgentClientDefinition { def apply[Constructor]( - componentId: ComponentId, name: String, constructor: InputRecordCodec[Constructor], mode: AgentMode = AgentMode.Durable ): AgentClientDefinition[Constructor] = - new AgentClientDefinition(componentId, name, mode, constructor) + new AgentClientDefinition(name, mode, constructor) } final case class CallerCodecMethod[Input, Output]( @@ -74,8 +89,10 @@ final class CallerCodecClientFactory[Constructor] private[reflection] ( val phantom = Uuid.random() for { constructor <- encodeConstructor(input) - id <- AgentId.create(definition.componentId, definition.name, constructor, Some(phantom)) - client <- createValue(constructor, Some(phantom)) + componentId <- Reflection.componentIdFor(definition.name) + id <- AgentId.create(componentId, definition.name, constructor, Some(phantom)) + transport <- Transport.create(componentId, definition.name, constructor, Some(phantom)) + client = new CallerCodecAgentClient(definition, transport) } yield Right(CallerCodecPhantomClient(id, phantom, client)) } @@ -89,8 +106,9 @@ final class CallerCodecClientFactory[Constructor] private[reflection] ( constructor: SchemaValue, phantomId: Option[Uuid] ): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = - Transport - .create(definition.componentId, definition.name, constructor, phantomId) + Reflection + .componentIdFor(definition.name) + .flatMap(Transport.create(_, definition.name, constructor, phantomId)) .map(new CallerCodecAgentClient(definition, _)) private def encodeConstructor(input: Constructor): Either[GolemReflectError, SchemaValue] = @@ -129,10 +147,12 @@ final class CallerCodecBoundMethod[Input, Output] private[reflection] ( } def trigger(input: Input): Either[GolemReflectError, InvocationMetadata] = - encodeInput(input).flatMap(transport.trigger(definition.name, _)) + rejectNonAwaitedStreams("trigger").flatMap(_ => encodeInput(input)).flatMap(transport.trigger(definition.name, _)) def schedule(at: Datetime, input: Input): Either[GolemReflectError, ScheduledInvocation] = - encodeInput(input).flatMap(transport.schedule(at, definition.name, _)) + rejectNonAwaitedStreams("schedule") + .flatMap(_ => encodeInput(input)) + .flatMap(transport.schedule(at, definition.name, _)) private def encodeInput(input: Input): Either[GolemReflectError, SchemaValue] = try Right(definition.input.toValue(input)) @@ -156,6 +176,18 @@ final class CallerCodecBoundMethod[Input, Output] private[reflection] ( .map(error => GolemReflectError.SchemaDecode(error.message)) ) } + + private def rejectNonAwaitedStreams(operation: String): Either[GolemReflectError, Unit] = { + val outputContainsStream = definition.output.metadata match { + case OutputMetadata.Unit => false + case OutputMetadata.Single(graph) => graph.containsStream + } + Either.cond( + !definition.input.graph.containsStream && !outputContainsStream, + (), + GolemReflectError.Validation(s"$operation is unavailable for streaming method '${definition.name}'") + ) + } } final case class TypedInvocation[+A](metadata: InvocationMetadata, value: A) diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala index b71a1187a7..d1e09701e8 100644 --- a/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala @@ -56,6 +56,9 @@ object ComponentId { final case class AgentId(componentId: ComponentId, value: String) { def parts: Either[GolemReflectError, AgentIdParts] = AgentId.parse(this) def dynamicClient: Either[GolemReflectError, DynamicAgentClient] = DynamicAgentClient.fromAgentId(this) + def client[Constructor]( + definition: AgentClientDefinition[Constructor] + ): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = definition.bind(this) } final case class AgentIdParts(typeName: String, constructorValue: SchemaValue, phantomId: Option[Uuid]) @@ -71,7 +74,7 @@ object AgentId { AgentHostApi .makeAgentId(typeName, payload, phantomId) .left - .map(GolemReflectError.Identity) + .map(GolemReflectError.Identity.apply) .map(AgentId(componentId, _)) ) @@ -79,7 +82,7 @@ object AgentId { AgentHostApi .parseAgentId(agentId.value) .left - .map(GolemReflectError.Identity) + .map(GolemReflectError.Identity.apply) .flatMap { parts => try Right( @@ -168,17 +171,26 @@ object Reflection { .fold[Either[GolemReflectError, Option[AgentType]]](Right(None))(_.map(Some(_))) catch { case NonFatal(error) => Left(GolemReflectError.Discovery(error.getMessage)) } + private[reflection] def componentIdFor(name: String): Either[GolemReflectError, ComponentId] = + try + AgentHostApi + .registeredAgentType(name) + .map(value => ComponentId.fromJs(value.implementedBy)) + .toRight(GolemReflectError.Discovery(s"Agent type '$name' is not registered in the current environment")) + catch { case NonFatal(error) => Left(GolemReflectError.Discovery(error.getMessage)) } + private def decodeAgentType(registered: AgentHostApi.RegisteredAgentType): Either[GolemReflectError, AgentType] = try { val raw = registered.agentType val graph = raw.schema + val decoded = SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph)) val methods = raw.methods.toList.map { method => AgentMethod( method.name, method.description, method.promptHint.toOption, - inputRef(graph, method.inputSchema), - outputRef(graph, method.outputSchema) + inputRef(graph, decoded, method.inputSchema), + outputRef(graph, decoded, method.outputSchema) ) } val mode = raw.mode match { @@ -193,15 +205,14 @@ object Reflection { raw.sourceLanguage, mode, ComponentId.fromJs(registered.implementedBy), - inputRef(graph, raw.constructor.inputSchema), + inputRef(graph, decoded, raw.constructor.inputSchema), methods ) ) } catch { case NonFatal(error) => Left(GolemReflectError.SchemaDecode(error.getMessage)) } - private def inputRef(graph: JsSchemaGraph, input: JsInputSchema): SchemaRef = { + private def inputRef(graph: JsSchemaGraph, decoded: SchemaGraph, input: JsInputSchema): SchemaRef = { if (input.tag != "parameters") throw new IllegalArgumentException(s"unknown input schema '${input.tag}'") - val decoded = SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph)) val entries = input.asInstanceOf[js.Dynamic].selectDynamic("val").asInstanceOf[js.Array[JsNamedField]].toList val fields = entries.collect { case entry if entry.source.tag == "user-supplied" => @@ -211,12 +222,13 @@ object Reflection { SchemaRef(SchemaGraph(decoded.defs, SchemaType(RecordType(fields)))) } - private def outputRef(graph: JsSchemaGraph, output: JsOutputSchema): Option[SchemaRef] = + private def outputRef(graph: JsSchemaGraph, decoded: SchemaGraph, output: JsOutputSchema): Option[SchemaRef] = output.tag match { case "unit" => None case "single" => - val root = output.asInstanceOf[js.Dynamic].selectDynamic("val").asInstanceOf[Int] - Some(SchemaRef(SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph).copy(root = root)))) + val root = output.asInstanceOf[js.Dynamic].selectDynamic("val").asInstanceOf[Int] + val rooted = SchemaWire.schemaGraphFromWit(SchemaWireInterop.graphFromJs(graph).copy(root = root)).root + Some(SchemaRef(decoded, rooted)) case other => throw new IllegalArgumentException(s"unknown output schema '$other'") } } @@ -300,11 +312,14 @@ final class ReflectedAgentMethod private[reflection] (val definition: AgentMetho def invokeValue(input: SchemaValue): Future[Either[GolemReflectError, Invocation[SchemaValue]]] = validate(definition.input, input) match { case Left(error) => Future.successful(Left(error)) - case Right(_) => transport.invokeAndAwait(definition.name, input) + case Right(_) => + transport.invokeAndAwait(definition.name, input).map(_.flatMap(validateInvocationOutput(definition, _))) } def triggerValue(input: SchemaValue): Either[GolemReflectError, InvocationMetadata] = - validate(definition.input, input).flatMap(_ => transport.trigger(definition.name, input)) + rejectNonAwaitedStreams("trigger") + .flatMap(_ => validate(definition.input, input)) + .flatMap(_ => transport.trigger(definition.name, input)) def triggerJson(input: Json): Either[GolemReflectError, InvocationMetadata] = definition.input @@ -314,7 +329,9 @@ final class ReflectedAgentMethod private[reflection] (val definition: AgentMetho .flatMap(triggerValue) def scheduleValue(at: Datetime, input: SchemaValue): Either[GolemReflectError, ScheduledInvocation] = - validate(definition.input, input).flatMap(_ => transport.schedule(at, definition.name, input)) + rejectNonAwaitedStreams("schedule") + .flatMap(_ => validate(definition.input, input)) + .flatMap(_ => transport.schedule(at, definition.name, input)) def scheduleJson(at: Datetime, input: Json): Either[GolemReflectError, ScheduledInvocation] = definition.input @@ -322,6 +339,13 @@ final class ReflectedAgentMethod private[reflection] (val definition: AgentMetho .left .map(error => GolemReflectError.Validation(error.message)) .flatMap(scheduleValue(at, _)) + + private def rejectNonAwaitedStreams(operation: String): Either[GolemReflectError, Unit] = + Either.cond( + !definition.input.containsStream && !definition.output.exists(_.containsStream), + (), + GolemReflectError.Validation(s"$operation is unavailable for streaming method '${definition.name}'") + ) } final case class InvocationMetadata(agentId: AgentId, idempotencyKey: String) @@ -409,7 +433,7 @@ private[reflection] object Transport { } } -private object ReflectionInternals { +private[reflection] object ReflectionInternals { def validate(schema: SchemaRef, value: SchemaValue): Either[GolemReflectError, Unit] = schema .validateValue(value) @@ -417,6 +441,17 @@ private object ReflectionInternals { .map(errors => GolemReflectError.Validation(errors.map(_.message).mkString("; "))) .map(_ => ()) + def validateInvocationOutput( + definition: AgentMethod, + invocation: Invocation[SchemaValue] + ): Either[GolemReflectError, Invocation[SchemaValue]] = + (definition.output, invocation.value) match { + case (None, None) => Right(invocation) + case (Some(schema), Some(value)) => validate(schema, value).map(_ => invocation) + case (None, Some(_)) => Left(GolemReflectError.SchemaDecode("unit method returned a value")) + case (Some(_), None) => Left(GolemReflectError.SchemaDecode("single-output method returned no value")) + } + def encode(value: SchemaValue): Either[GolemReflectError, JsSchemaValueTree] = try Right(SchemaWireInterop.valueTreeToJs(SchemaWire.schemaValueToWit(value))) catch { case NonFatal(error) => Left(GolemReflectError.SchemaEncode(error.getMessage)) } diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala index 05c47312d1..553185c9dc 100644 --- a/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/SchemaRef.scala @@ -21,6 +21,8 @@ import scala.collection.immutable.ListMap final case class SchemaIssue(message: String, path: List[String] = Nil) final class SchemaRef private (val graph: SchemaGraph, val root: SchemaType) { + def containsStream: Boolean = graph.containsStream + def validateValue(value: SchemaValue): Either[List[SchemaIssue], SchemaValue] = ValueValidation .validateValue(graph, root, value) diff --git a/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala b/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala index 3293b06cc4..e0e8d9f882 100644 --- a/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala +++ b/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala @@ -10,6 +10,7 @@ package golem.reflection +import golem.Uuid import golem.schema._ import golem.schema.SchemaTypeBody._ import golem.schema.SchemaValue._ @@ -97,6 +98,32 @@ object SchemaRefSpec extends ZIOSpecDefault { rendered.get("$schema").one == Right(Json.String("https://json-schema.org/draft/2020-12/schema")), rendered.get("type").one == Right(Json.String("object")) ) + }, + test("detects nested stream schemas") { + val streaming = SchemaRef( + SchemaGraph( + ListMap.empty, + SchemaType(RecordType(List(NamedFieldType("items", SchemaType(StreamType(Some(SchemaType(StringType)))))))) + ) + ) + assertTrue(!SchemaRef(graph).containsStream, streaming.containsStream) + }, + test("rejects missing, unexpected, and malformed reflected outputs") { + val input = SchemaRef(SchemaGraph(ListMap.empty, SchemaType(RecordType(Nil)))) + val output = SchemaRef(SchemaGraph(ListMap.empty, SchemaType(StringType))) + val metadata = InvocationMetadata(AgentId(ComponentId(Uuid(0, 0)), "test"), "key") + val unit = AgentMethod("unit", "", None, input, None) + val single = AgentMethod("single", "", None, input, Some(output)) + + assertTrue( + ReflectionInternals.validateInvocationOutput(unit, Invocation(metadata, None)).isRight, + ReflectionInternals.validateInvocationOutput(unit, Invocation(metadata, Some(StringValue("extra")))).isLeft, + ReflectionInternals.validateInvocationOutput(single, Invocation(metadata, None)).isLeft, + ReflectionInternals.validateInvocationOutput(single, Invocation(metadata, Some(U32Value(1)))).isLeft, + ReflectionInternals + .validateInvocationOutput(single, Invocation(metadata, Some(StringValue("ok")))) + .isRight + ) } ) } From 2e7047224b2b3e3146e8b51234b559aeae6c596e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Mon, 7 Sep 2026 11:57:58 +0200 Subject: [PATCH 3/4] Load common reflection guidance for Scala --- .../tests/harness/scenarios/rpc-5-runtime-reflection.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml b/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml index 0f065000d1..b604bc070a 100644 --- a/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml +++ b/golem-skills/tests/harness/scenarios/rpc-5-runtime-reflection.yaml @@ -165,6 +165,7 @@ steps: `methodName`, `jsonValue`, `callerCodec`, `reflectedValue`, and `directValue`. Make sure the project builds successfully. expectedSkills: + - "golem-agent-reflection" - "golem-agent-reflection-scala" verify: build: true From 92545f2350a5a4278f0b8a49df9bf99b3c9c9f74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Thu, 10 Sep 2026 15:09:27 +0200 Subject: [PATCH 4/4] Use environment-scoped IDs in Scala reflection --- .../scala/golem-agent-reflection-scala.mdx | 11 ++-- .../golem-agent-reflection-scala/SKILL.md | 11 ++-- .../golem/reflection/CallerCodecClient.scala | 17 +++--- .../scala/golem/reflection/Reflection.scala | 55 +++++++++---------- .../golem/runtime/rpc/host/AgentHostApi.scala | 9 ++- .../golem/reflection/SchemaRefSpec.scala | 3 +- 6 files changed, 52 insertions(+), 54 deletions(-) diff --git a/docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx b/docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx index 24d8a6d074..5696d68a20 100644 --- a/docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx +++ b/docs/src/content/next/how-to-guides/scala/golem-agent-reflection-scala.mdx @@ -66,7 +66,7 @@ stream. `AgentClientDefinition` does not discover remote schemas. `InputRecordCodec` and `OutputCodec` are the caller's schema authority; the environment-unique -type name is used only to resolve current implementation identity metadata: +type name is the invocation address: ```scala import golem.reflection._ @@ -87,17 +87,16 @@ val result = counter.map(_.method(add).invoke(5)) ``` Use `agentId.client(contract)` to bind the same caller-owned codecs to an -existing durable identity. Both creation and binding fail when the named type -is not registered in the current environment. +existing durable identity. ## Lifecycle attempts -- Use a supplied `AgentId` directly or inspect it with `parts`. -- Use `AgentId.create` for schema-free durable, known-phantom, or newly +- Use a supplied `ParsedAgentId` directly or inspect it with `parts`. +- Use `ParsedAgentId.create` for schema-free durable, known-phantom, or newly generated phantom identities. - Reflected and caller-codec factories provide `get`, `getPhantom`, and `newPhantom`. -- Use `DynamicAgentClient.ephemeral(componentId, typeName, constructorValue)` +- Use `DynamicAgentClient.ephemeral(typeName, constructorValue)` for a raw ephemeral invocation address. An ephemeral address has no guaranteed reusable pre-invocation identity. The diff --git a/golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md b/golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md index 0ee1655306..649f23d812 100644 --- a/golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md +++ b/golem-skills/skills/scala/golem-agent-reflection-scala/SKILL.md @@ -71,7 +71,7 @@ stream. `AgentClientDefinition` does not discover remote schemas. `InputRecordCodec` and `OutputCodec` are the caller's schema authority; the environment-unique -type name is used only to resolve current implementation identity metadata: +type name is the invocation address: ```scala import golem.reflection._ @@ -92,17 +92,16 @@ val result = counter.map(_.method(add).invoke(5)) ``` Use `agentId.client(contract)` to bind the same caller-owned codecs to an -existing durable identity. Both creation and binding fail when the named type -is not registered in the current environment. +existing durable identity. ## Lifecycle attempts -- Use a supplied `AgentId` directly or inspect it with `parts`. -- Use `AgentId.create` for schema-free durable, known-phantom, or newly +- Use a supplied `ParsedAgentId` directly or inspect it with `parts`. +- Use `ParsedAgentId.create` for schema-free durable, known-phantom, or newly generated phantom identities. - Reflected and caller-codec factories provide `get`, `getPhantom`, and `newPhantom`. -- Use `DynamicAgentClient.ephemeral(componentId, typeName, constructorValue)` +- Use `DynamicAgentClient.ephemeral(typeName, constructorValue)` for a raw ephemeral invocation address. An ephemeral address has no guaranteed reusable pre-invocation identity. The diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala index 477a6962a6..dadc411ad7 100644 --- a/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/CallerCodecClient.scala @@ -33,7 +33,7 @@ final class AgentClientDefinition[Constructor] private ( ): CallerCodecMethod[Input, Output] = CallerCodecMethod(name, input, output) - def bind(agentId: AgentId): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = + def bind(agentId: ParsedAgentId): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = for { parts <- agentId.parts _ <- Either.cond( @@ -46,8 +46,7 @@ final class AgentClientDefinition[Constructor] private ( (), GolemReflectError.Identity(s"Cannot bind an existing identity to ephemeral agent type '$name'") ) - componentId <- Reflection.componentIdFor(name) - transport <- Transport.create(componentId, name, parts.constructorValue, parts.phantomId) + transport <- Transport.create(name, parts.constructorValue, parts.phantomId) } yield new CallerCodecAgentClient(this, transport) } @@ -67,7 +66,7 @@ final case class CallerCodecMethod[Input, Output]( ) final case class CallerCodecPhantomClient[Constructor]( - agentId: AgentId, + agentId: ParsedAgentId, phantomId: Uuid, client: CallerCodecAgentClient[Constructor] ) @@ -89,9 +88,8 @@ final class CallerCodecClientFactory[Constructor] private[reflection] ( val phantom = Uuid.random() for { constructor <- encodeConstructor(input) - componentId <- Reflection.componentIdFor(definition.name) - id <- AgentId.create(componentId, definition.name, constructor, Some(phantom)) - transport <- Transport.create(componentId, definition.name, constructor, Some(phantom)) + id <- ParsedAgentId.create(definition.name, constructor, Some(phantom)) + transport <- Transport.create(definition.name, constructor, Some(phantom)) client = new CallerCodecAgentClient(definition, transport) } yield Right(CallerCodecPhantomClient(id, phantom, client)) } @@ -106,9 +104,8 @@ final class CallerCodecClientFactory[Constructor] private[reflection] ( constructor: SchemaValue, phantomId: Option[Uuid] ): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = - Reflection - .componentIdFor(definition.name) - .flatMap(Transport.create(_, definition.name, constructor, phantomId)) + Transport + .create(definition.name, constructor, phantomId) .map(new CallerCodecAgentClient(definition, _)) private def encodeConstructor(input: Constructor): Either[GolemReflectError, SchemaValue] = diff --git a/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala b/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala index d1e09701e8..96002ea4be 100644 --- a/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala +++ b/sdks/scala/core/js/src/main/scala/golem/reflection/Reflection.scala @@ -53,32 +53,31 @@ object ComponentId { ComponentId(Uuid(BigInt(value.uuid.highBits.toString), BigInt(value.uuid.lowBits.toString))) } -final case class AgentId(componentId: ComponentId, value: String) { - def parts: Either[GolemReflectError, AgentIdParts] = AgentId.parse(this) +final case class ParsedAgentId(value: String) { + def parts: Either[GolemReflectError, ParsedAgentIdParts] = ParsedAgentId.parse(this) def dynamicClient: Either[GolemReflectError, DynamicAgentClient] = DynamicAgentClient.fromAgentId(this) def client[Constructor]( definition: AgentClientDefinition[Constructor] ): Either[GolemReflectError, CallerCodecAgentClient[Constructor]] = definition.bind(this) } -final case class AgentIdParts(typeName: String, constructorValue: SchemaValue, phantomId: Option[Uuid]) +final case class ParsedAgentIdParts(typeName: String, constructorValue: SchemaValue, phantomId: Option[Uuid]) -object AgentId { +object ParsedAgentId { def create( - componentId: ComponentId, typeName: String, constructorValue: SchemaValue, phantomId: Option[Uuid] = None - ): Either[GolemReflectError, AgentId] = + ): Either[GolemReflectError, ParsedAgentId] = encode(constructorValue).flatMap(payload => AgentHostApi .makeAgentId(typeName, payload, phantomId) .left .map(GolemReflectError.Identity.apply) - .map(AgentId(componentId, _)) + .map(ParsedAgentId(_)) ) - def parse(agentId: AgentId): Either[GolemReflectError, AgentIdParts] = + def parse(agentId: ParsedAgentId): Either[GolemReflectError, ParsedAgentIdParts] = AgentHostApi .parseAgentId(agentId.value) .left @@ -86,7 +85,7 @@ object AgentId { .flatMap { parts => try Right( - AgentIdParts( + ParsedAgentIdParts( parts.agentTypeName, SchemaWire.schemaValueFromWit(SchemaWireInterop.valueTreeFromJs(parts.payload.value)), parts.phantom @@ -131,17 +130,17 @@ final class AgentType private[reflection] ( def method(name: String): Option[AgentMethod] = methods.find(_.name == name) - def agentId(input: Json, phantomId: Option[Uuid] = None): Either[GolemReflectError, AgentId] = + def agentId(input: Json, phantomId: Option[Uuid] = None): Either[GolemReflectError, ParsedAgentId] = constructorInput .packJson(input) .left .map(error => GolemReflectError.Validation(error.message)) .flatMap(agentIdValue(_, phantomId)) - def agentIdValue(input: SchemaValue, phantomId: Option[Uuid] = None): Either[GolemReflectError, AgentId] = - validate(constructorInput, input).flatMap(_ => AgentId.create(implementedBy, name, input, phantomId)) + def agentIdValue(input: SchemaValue, phantomId: Option[Uuid] = None): Either[GolemReflectError, ParsedAgentId] = + validate(constructorInput, input).flatMap(_ => ParsedAgentId.create(name, input, phantomId)) - def bind(agentId: AgentId): Either[GolemReflectError, ReflectedAgentClient] = + def bind(agentId: ParsedAgentId): Either[GolemReflectError, ReflectedAgentClient] = for { parts <- agentId.parts _ <- Either.cond( @@ -171,12 +170,12 @@ object Reflection { .fold[Either[GolemReflectError, Option[AgentType]]](Right(None))(_.map(Some(_))) catch { case NonFatal(error) => Left(GolemReflectError.Discovery(error.getMessage)) } - private[reflection] def componentIdFor(name: String): Either[GolemReflectError, ComponentId] = + def getAgentTypeFor(agentId: ParsedAgentId): Either[GolemReflectError, Option[AgentType]] = try AgentHostApi - .registeredAgentType(name) - .map(value => ComponentId.fromJs(value.implementedBy)) - .toRight(GolemReflectError.Discovery(s"Agent type '$name' is not registered in the current environment")) + .registeredAgentTypeFor(agentId.value) + .map(decodeAgentType) + .fold[Either[GolemReflectError, Option[AgentType]]](Right(None))(_.map(Some(_))) catch { case NonFatal(error) => Left(GolemReflectError.Discovery(error.getMessage)) } private def decodeAgentType(registered: AgentHostApi.RegisteredAgentType): Either[GolemReflectError, AgentType] = @@ -233,7 +232,7 @@ object Reflection { } } -final case class ReflectedPhantomClient(agentId: AgentId, phantomId: Uuid, client: ReflectedAgentClient) +final case class ReflectedPhantomClient(agentId: ParsedAgentId, phantomId: Uuid, client: ReflectedAgentClient) final class ReflectedAgentClientFactory private[reflection] (agentType: AgentType) { def get(input: Json): Either[GolemReflectError, ReflectedAgentClient] = @@ -268,7 +267,7 @@ final class ReflectedAgentClientFactory private[reflection] (agentType: AgentTyp phantomId: Option[Uuid] ): Either[GolemReflectError, ReflectedAgentClient] = validate(agentType.constructorInput, input) - .flatMap(_ => Transport.create(agentType.implementedBy, agentType.name, input, phantomId)) + .flatMap(_ => Transport.create(agentType.name, input, phantomId)) .map(new ReflectedAgentClient(agentType, _)) private def pack(input: Json): Either[GolemReflectError, SchemaValue] = @@ -348,29 +347,28 @@ final class ReflectedAgentMethod private[reflection] (val definition: AgentMetho ) } -final case class InvocationMetadata(agentId: AgentId, idempotencyKey: String) +final case class InvocationMetadata(agentId: ParsedAgentId, idempotencyKey: String) final case class Invocation[+A](metadata: InvocationMetadata, value: Option[A]) final case class ScheduledInvocation(metadata: InvocationMetadata, cancellationToken: CancellationToken) -final class DynamicAgentClient private (transport: Transport, val agentId: Option[AgentId]) { +final class DynamicAgentClient private (transport: Transport, val agentId: Option[ParsedAgentId]) { def method(name: String): DynamicAgentMethod = new DynamicAgentMethod(name, transport) } object DynamicAgentClient { - def fromAgentId(agentId: AgentId): Either[GolemReflectError, DynamicAgentClient] = + def fromAgentId(agentId: ParsedAgentId): Either[GolemReflectError, DynamicAgentClient] = agentId.parts - .flatMap(parts => Transport.create(agentId.componentId, parts.typeName, parts.constructorValue, parts.phantomId)) + .flatMap(parts => Transport.create(parts.typeName, parts.constructorValue, parts.phantomId)) .map(new DynamicAgentClient(_, Some(agentId))) /** * A raw one-shot address. Final identity is supplied by invocation metadata. */ def ephemeral( - componentId: ComponentId, typeName: String, constructor: SchemaValue ): Either[GolemReflectError, DynamicAgentClient] = - Transport.create(componentId, typeName, constructor, None).map(new DynamicAgentClient(_, None)) + Transport.create(typeName, constructor, None).map(new DynamicAgentClient(_, None)) } final class DynamicAgentMethod private[reflection] (val name: String, transport: Transport) { @@ -381,7 +379,7 @@ final class DynamicAgentMethod private[reflection] (val name: String, transport: transport.schedule(at, name, input) } -private[reflection] final class Transport private (componentId: ComponentId, raw: WasmRpcApi.WasmRpcClient) { +private[reflection] final class Transport private (raw: WasmRpcApi.WasmRpcClient) { def invokeAndAwait(method: String, input: SchemaValue): Future[Either[GolemReflectError, Invocation[SchemaValue]]] = encodeAsync(input).flatMap { payload => raw.asyncInvokeAndAwaitWithMetadata(method, payload) match { @@ -415,12 +413,11 @@ private[reflection] final class Transport private (componentId: ComponentId, raw ) private def toMetadata(value: golem.runtime.rpc.InvocationMetadata): InvocationMetadata = - InvocationMetadata(AgentId(componentId, value.agentId), value.idempotencyKey) + InvocationMetadata(ParsedAgentId(value.agentId), value.idempotencyKey) } private[reflection] object Transport { def create( - componentId: ComponentId, typeName: String, constructor: SchemaValue, phantom: Option[Uuid] @@ -429,7 +426,7 @@ private[reflection] object Transport { val phantomArg = phantom.fold[js.UndefOr[JsSchemaUuid]](js.undefined)(uuid => JsSchemaUuid(js.BigInt(uuid.highBits.toString), js.BigInt(uuid.lowBits.toString)) ) - new Transport(componentId, WasmRpcApi.newClient(typeName, payload, phantomArg, js.Array())) + new Transport(WasmRpcApi.newClient(typeName, payload, phantomArg, js.Array())) } } diff --git a/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/AgentHostApi.scala b/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/AgentHostApi.scala index 76826d1ffc..ee373b787e 100644 --- a/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/AgentHostApi.scala +++ b/sdks/scala/core/js/src/main/scala/golem/runtime/rpc/host/AgentHostApi.scala @@ -76,6 +76,11 @@ object AgentHostApi { if (v == null || js.isUndefined(v)) None else Some(v) } + def registeredAgentTypeFor(agentId: String): Option[RegisteredAgentType] = { + val value = AgentRegistryModule.getAgentTypeByAgentId(agentId) + if (value == null || js.isUndefined(value)) None else Some(value) + } + def getAllAgentTypes(): List[RegisteredAgentType] = AgentRegistryModule.getAllAgentTypes().toList @@ -138,7 +143,7 @@ object AgentHostApi { AgentRegistryModule.createWebhook(promiseId) def getConfigValue(key: List[String], expected: JsSchemaGraph): JsSchemaValueTree = - AgentRegistryModule.getConfigValue(js.Array(key: _*), expected) + AgentRegistryModule.getConfigValue(js.Array(key*), expected) def getOplogIndex(): OplogIndex = HostModule.getOplogIndex() @@ -387,6 +392,8 @@ object AgentHostApi { def getAllAgentTypes(): js.Array[RegisteredAgentType] = js.native + def getAgentTypeByAgentId(agentId: String): RegisteredAgentType = js.native + def makeAgentId(agentTypeName: String, input: JsSchemaValueTree, phantom: js.Any): String = js.native def parseAgentId(agentId: String): js.Array[js.Any] = js.native diff --git a/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala b/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala index e0e8d9f882..49992e4eaf 100644 --- a/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala +++ b/sdks/scala/core/js/src/test/scala/golem/reflection/SchemaRefSpec.scala @@ -10,7 +10,6 @@ package golem.reflection -import golem.Uuid import golem.schema._ import golem.schema.SchemaTypeBody._ import golem.schema.SchemaValue._ @@ -111,7 +110,7 @@ object SchemaRefSpec extends ZIOSpecDefault { test("rejects missing, unexpected, and malformed reflected outputs") { val input = SchemaRef(SchemaGraph(ListMap.empty, SchemaType(RecordType(Nil)))) val output = SchemaRef(SchemaGraph(ListMap.empty, SchemaType(StringType))) - val metadata = InvocationMetadata(AgentId(ComponentId(Uuid(0, 0)), "test"), "key") + val metadata = InvocationMetadata(ParsedAgentId("test"), "key") val unit = AgentMethod("unit", "", None, input, None) val single = AgentMethod("single", "", None, input, Some(output))