diff --git a/composer.json b/composer.json
index d3ea5fb04..4bc6543b5 100644
--- a/composer.json
+++ b/composer.json
@@ -30,6 +30,7 @@
"internal/destroy": "^1.0",
"internal/promise": "^3.4",
"nesbot/carbon": "^2.72.6 || ^3.8.4",
+ "psr/clock": "^1.0",
"psr/log": "^2.0 || ^3.0.2",
"ramsey/uuid": "^4.7.6",
"roadrunner-php/roadrunner-api-dto": "^1.14.0",
diff --git a/phpunit.xml.dist b/phpunit.xml.dist
index f1ee28874..8d55fca81 100644
--- a/phpunit.xml.dist
+++ b/phpunit.xml.dist
@@ -75,6 +75,7 @@
tests/Unit
+ tests/Nexus/Unit
tests/Functional
diff --git a/psalm-baseline.xml b/psalm-baseline.xml
index 36b6ddf16..a8e764d1f 100644
--- a/psalm-baseline.xml
+++ b/psalm-baseline.xml
@@ -31,11 +31,6 @@
-
-
-
-
-
name]]>
@@ -99,9 +94,6 @@
excludeCalendarList]]>
excludeCalendarList]]>
-
-
-
@@ -258,11 +250,6 @@
-
-
-
-
-
@@ -447,9 +434,6 @@
-
- ]]>
-
]]>
@@ -515,6 +499,12 @@
+ ]]>
+
+
+
+
+ ]]>
@@ -533,7 +523,8 @@
- ]]>
+ ]]>
+ ]]>
@@ -585,18 +576,6 @@
-
-
-
-
-
-
-
-
-
-
-
-
getRunId()]]>
@@ -609,16 +588,6 @@
-
-
-
-
-
-
-
-
-
-
@@ -817,17 +786,7 @@
-
-
-
-
-
-
-
-
-
-
@@ -924,10 +883,6 @@
-
-
-
-
@@ -936,13 +891,6 @@
getMessage()]]>
-
-
-
-
-
-
-
@@ -1003,6 +951,17 @@
+
+
+
+
+
+
+
+
+ operation]]>
+
+
@@ -1060,6 +1019,7 @@
maxSupported]]>
minSupported]]>
+ operation]]>
type]]>
@@ -1102,6 +1062,37 @@
trace = &$this->trace]]>
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ workflowId]]>
+
+
@@ -1182,12 +1173,6 @@
-
- serializeToString()]]>
-
-
-
-
getCode()]]>
getCode()]]>
@@ -1210,22 +1195,10 @@
-
- getOptions(), JSON_INVALID_UTF8_IGNORE | JSON_UNESCAPED_UNICODE)]]>
-
-
getFailure()]]>
-
-
-
-
-
- failure]]>
-
-
@@ -1360,6 +1333,12 @@
+
+
+
+
+
+
@@ -1492,19 +1471,6 @@
getHeader()]]>
getPayloads()]]>
-
-
-
-
-
-
-
- getSeconds() + \round($eventTime->getNanos() / 1_000_000_000, 6)]]>
-
-
-
- getSeconds()]]>
-
diff --git a/psalm.xml b/psalm.xml
index ef14c51ce..9c160f6d4 100644
--- a/psalm.xml
+++ b/psalm.xml
@@ -22,6 +22,11 @@
+
+
+
+
+
diff --git a/src/Client/WorkflowOptions.php b/src/Client/WorkflowOptions.php
index 027f2c522..a3fdd7b55 100644
--- a/src/Client/WorkflowOptions.php
+++ b/src/Client/WorkflowOptions.php
@@ -31,10 +31,14 @@
use Temporal\Internal\Marshaller\Type\CronType;
use Temporal\Internal\Marshaller\Type\DateIntervalType;
use Temporal\Internal\Marshaller\Type\NullableType;
+use Temporal\Internal\Nexus\NexusLinkConverter;
+use Temporal\Internal\Client\OnConflictOptions;
use Temporal\Internal\Support\DateInterval;
use Temporal\Internal\Support\Options;
+use Temporal\Nexus\Link as NexusLink;
use Temporal\Worker\Worker;
use Temporal\Worker\WorkerFactoryInterface;
+use Temporal\Workflow\CompletionCallback;
/**
* WorkflowOptions configuration parameters for starting a workflow execution.
@@ -193,6 +197,17 @@ final class WorkflowOptions extends Options
#[Marshal(name: 'VersioningOverride')]
public ?VersioningOverride $versioningOverride = null;
+ public ?string $requestId = null;
+
+ /** @var list */
+ public array $completionCallbacks = [];
+
+ /** @internal */
+ public ?OnConflictOptions $onConflictOptions = null;
+
+ /** @var list<\Temporal\Api\Common\V1\Link> */
+ public array $links = [];
+
/**
* @throws \Exception
*/
@@ -612,4 +627,109 @@ public function withPriority(Priority $priority): self
$self->priority = $priority;
return $self;
}
+
+ /**
+ * Pin gRPC `request_id`. `null` = fresh UUID per call.
+ *
+ * @return $this
+ */
+ #[Pure]
+ public function withRequestId(?string $requestId): self
+ {
+ $self = clone $this;
+ $self->requestId = $requestId;
+ return $self;
+ }
+
+ /**
+ * Append a Nexus completion callback. Use {@see self::withCompletionCallbacks()}
+ * with an empty argument list to clear.
+ *
+ * @param non-empty-string $url
+ * @param array $headers
+ * @return $this
+ */
+ #[Pure]
+ public function withNexusCompletionCallback(string $url, array $headers = []): self
+ {
+ $self = clone $this;
+ $self->completionCallbacks = [...$this->completionCallbacks, new CompletionCallback($url, $headers)];
+ return $self;
+ }
+
+ /**
+ * Replace the full list of completion callbacks.
+ *
+ * @return $this
+ */
+ #[Pure]
+ public function withCompletionCallbacks(CompletionCallback ...$callbacks): self
+ {
+ $self = clone $this;
+ $self->completionCallbacks = \array_values($callbacks);
+ return $self;
+ }
+
+ /**
+ * @internal
+ * @return $this
+ */
+ #[Pure]
+ public function withOnConflictOptionsInternal(?OnConflictOptions $options): self
+ {
+ $self = clone $this;
+ $self->onConflictOptions = $options;
+ return $self;
+ }
+
+ /**
+ * Replace the link list with proto Link[] derived from Nexus-level URIs.
+ *
+ * @psalm-suppress ImpureMethodCall
+ *
+ * @param iterable $nexusLinks
+ * @return $this
+ */
+ #[Pure]
+ public function withLinks(iterable $nexusLinks): self
+ {
+ $self = clone $this;
+ $self->links = NexusLinkConverter::toProtoLinks($nexusLinks);
+ return $self;
+ }
+
+ /**
+ * Snapshot of the DTO for `var_dump`/`print_r`/Xdebug pretty-printing.
+ *
+ * Renders {@see \DateInterval} fields as ISO 8601 spec strings (e.g.
+ * `PT60S`); the default dump expands them into nine zero components and
+ * buries the actual values.
+ */
+ public function __debugInfo(): array
+ {
+ return [
+ 'workflowId' => $this->workflowId,
+ 'taskQueue' => $this->taskQueue,
+ 'eagerStart' => $this->eagerStart,
+ 'workflowExecutionTimeout' => CarbonInterval::instance($this->workflowExecutionTimeout)->spec(),
+ 'workflowRunTimeout' => CarbonInterval::instance($this->workflowRunTimeout)->spec(),
+ 'workflowStartDelay' => CarbonInterval::instance($this->workflowStartDelay)->spec(),
+ 'workflowTaskTimeout' => CarbonInterval::instance($this->workflowTaskTimeout)->spec(),
+ 'workflowIdReusePolicy' => $this->workflowIdReusePolicy,
+ 'workflowIdConflictPolicy' => $this->workflowIdConflictPolicy,
+ 'retryOptions' => $this->retryOptions,
+ 'cronSchedule' => $this->cronSchedule,
+ 'memo' => $this->memo,
+ 'searchAttributes' => $this->searchAttributes,
+ 'typedSearchAttributes' => $this->typedSearchAttributes,
+ 'staticDetails' => $this->staticDetails,
+ 'staticSummary' => $this->staticSummary,
+ 'priority' => $this->priority,
+ 'versioningOverride' => $this->versioningOverride,
+ 'requestId' => $this->requestId,
+ 'completionCallbacks' => $this->completionCallbacks,
+ 'onConflictOptions' => $this->onConflictOptions,
+ 'links' => $this->links,
+ ];
+ }
}
diff --git a/src/DataConverter/EncodedValues.php b/src/DataConverter/EncodedValues.php
index c94ca3219..0d32ebfe6 100644
--- a/src/DataConverter/EncodedValues.php
+++ b/src/DataConverter/EncodedValues.php
@@ -60,6 +60,24 @@ public static function fromPayloads(Payloads $payloads, DataConverterInterface $
return static::fromPayloadCollection($payloads->getPayloads(), $dataConverter);
}
+ public static function fromPayload(?Payload $payload, DataConverterInterface $dataConverter): EncodedValues
+ {
+ $payloads = new Payloads();
+ if ($payload !== null) {
+ $payloads->setPayloads([$payload]);
+ }
+ return static::fromPayloads($payloads, $dataConverter);
+ }
+
+ public static function firstPayload(ValuesInterface $values): ?Payload
+ {
+ $payloads = $values->toPayloads()->getPayloads();
+ if ($payloads->count() === 0) {
+ return null;
+ }
+ return $payloads[0];
+ }
+
public static function sliceValues(
DataConverterInterface $converter,
ValuesInterface $values,
diff --git a/src/Exception/Client/MultyOperation/OperationStatus.php b/src/Exception/Client/MultyOperation/OperationStatus.php
index fea2e023e..8b191ccd1 100644
--- a/src/Exception/Client/MultyOperation/OperationStatus.php
+++ b/src/Exception/Client/MultyOperation/OperationStatus.php
@@ -17,7 +17,7 @@ final class OperationStatus
use UnpackDetailsTrait;
/**
- * @param \ArrayAccess&RepeatedField $details
+ * @param RepeatedField $details
*/
private function __construct(
private readonly \Traversable $details,
@@ -35,7 +35,7 @@ public function getMessage(): string
}
/**
- * @return \ArrayAccess&RepeatedField
+ * @return RepeatedField
*/
private function getDetails(): \Traversable
{
diff --git a/src/Exception/Client/ServiceClientException.php b/src/Exception/Client/ServiceClientException.php
index d92d81a10..fe54cd108 100644
--- a/src/Exception/Client/ServiceClientException.php
+++ b/src/Exception/Client/ServiceClientException.php
@@ -44,7 +44,7 @@ public function getStatus(): Status
}
/**
- * @return RepeatedField
+ * @return RepeatedField<\Google\Protobuf\Any>
*/
public function getDetails(): \ArrayAccess&\Countable&\IteratorAggregate
{
diff --git a/src/Exception/Client/UnpackDetailsTrait.php b/src/Exception/Client/UnpackDetailsTrait.php
index 82449f4bb..ce79059b4 100644
--- a/src/Exception/Client/UnpackDetailsTrait.php
+++ b/src/Exception/Client/UnpackDetailsTrait.php
@@ -31,7 +31,6 @@ public function getFailure(string $class): ?object
// ensures that message descriptor was added to the pool
Message::initOnce();
- /** @var Any $detail */
foreach ($details as $detail) {
if ($detail->is($class)) {
return $detail->unpack();
@@ -42,7 +41,7 @@ public function getFailure(string $class): ?object
}
/**
- * @return \ArrayAccess&RepeatedField
+ * @return RepeatedField
*/
abstract private function getDetails(): iterable;
}
diff --git a/src/Exception/Failure/FailureConverter.php b/src/Exception/Failure/FailureConverter.php
index 646cef035..04b67f7cf 100644
--- a/src/Exception/Failure/FailureConverter.php
+++ b/src/Exception/Failure/FailureConverter.php
@@ -11,6 +11,9 @@
namespace Temporal\Exception\Failure;
+use Temporal\Nexus\Exception\HandlerException as NexusHandlerException;
+use Temporal\Nexus\Exception\OperationException as NexusOperationException;
+use Temporal\Nexus\Internal\Failure\NexusFailureConverter;
use Temporal\Api\Common\V1\ActivityType;
use Temporal\Api\Common\V1\WorkflowExecution;
use Temporal\Api\Common\V1\WorkflowType;
@@ -19,6 +22,8 @@
use Temporal\Api\Failure\V1\CanceledFailureInfo;
use Temporal\Api\Failure\V1\ChildWorkflowExecutionFailureInfo;
use Temporal\Api\Failure\V1\Failure;
+use Temporal\Api\Failure\V1\NexusHandlerFailureInfo;
+use Temporal\Api\Failure\V1\NexusOperationFailureInfo;
use Temporal\Api\Failure\V1\ServerFailureInfo;
use Temporal\Api\Failure\V1\TerminatedFailureInfo;
use Temporal\Api\Failure\V1\TimeoutFailureInfo;
@@ -29,6 +34,8 @@
final class FailureConverter
{
+ public const NEXUS_OPERATION_ERROR_TYPE_PREFIX = 'nexus.OperationError.';
+
public static function mapFailureToException(Failure $failure, DataConverterInterface $converter): TemporalFailure
{
$e = self::createFailureException($failure, $converter);
@@ -148,6 +155,45 @@ public static function mapExceptionToFailure(\Throwable $e, DataConverterInterfa
$failure->setCanceledFailureInfo(new CanceledFailureInfo());
break;
+ case $e instanceof NexusHandlerException:
+ $info = new NexusHandlerFailureInfo();
+ $info->setType($e->errorType->value);
+ $info->setRetryBehavior(NexusFailureConverter::mapRetryBehavior($e->retryBehavior));
+
+ $failure->setNexusHandlerFailureInfo($info);
+ break;
+
+ case $e instanceof NexusHandlerFailure:
+ $info = new NexusHandlerFailureInfo();
+ $info->setType($e->getType());
+ $info->setRetryBehavior($e->getRetryBehavior());
+
+ $failure->setNexusHandlerFailureInfo($info);
+ break;
+
+ case $e instanceof NexusOperationException:
+ // Encode state in tagged ApplicationFailureInfo (no dedicated proto yet).
+ $info = new ApplicationFailureInfo();
+ $info->setType(self::NEXUS_OPERATION_ERROR_TYPE_PREFIX . $e->state->value);
+ $info->setNonRetryable(true);
+
+ $failure->setApplicationFailureInfo($info);
+ break;
+
+ case $e instanceof NexusOperationFailure:
+ $info = new NexusOperationFailureInfo();
+ /** @psalm-suppress DeprecatedMethod */
+ $info
+ ->setScheduledEventId($e->getScheduledEventId())
+ ->setEndpoint($e->getEndpoint())
+ ->setService($e->getService())
+ ->setOperation($e->getOperation())
+ ->setOperationId($e->getOperationToken())
+ ->setOperationToken($e->getOperationToken());
+
+ $failure->setNexusOperationExecutionFailureInfo($info);
+ break;
+
default:
$info = new ApplicationFailureInfo();
$info->setType($e::class);
@@ -261,6 +307,38 @@ private static function createFailureException(Failure $failure, DataConverterIn
$previous,
);
+ case $failure->hasNexusHandlerFailureInfo():
+ $info = $failure->getNexusHandlerFailureInfo();
+ \assert($info instanceof NexusHandlerFailureInfo);
+
+ return new NexusHandlerFailure(
+ $failure->getMessage(),
+ $info->getType(),
+ $info->getRetryBehavior(),
+ $previous,
+ );
+
+ case $failure->hasNexusOperationExecutionFailureInfo():
+ $info = $failure->getNexusOperationExecutionFailureInfo();
+ \assert($info instanceof NexusOperationFailureInfo);
+
+ // Fall back to deprecated operation_id for older servers.
+ $token = $info->getOperationToken();
+ if ($token === '') {
+ /** @psalm-suppress DeprecatedMethod */
+ $token = $info->getOperationId();
+ }
+
+ return new NexusOperationFailure(
+ $failure->getMessage(),
+ (int) $info->getScheduledEventId(),
+ $info->getEndpoint(),
+ $info->getService(),
+ $info->getOperation(),
+ $token,
+ $previous,
+ );
+
default:
throw new \InvalidArgumentException('Failure info not set');
}
diff --git a/src/Exception/Failure/NexusHandlerFailure.php b/src/Exception/Failure/NexusHandlerFailure.php
new file mode 100644
index 000000000..485d27929
--- /dev/null
+++ b/src/Exception/Failure/NexusHandlerFailure.php
@@ -0,0 +1,62 @@
+type;
+ }
+
+ /**
+ * {@see \Temporal\Api\Enums\V1\NexusHandlerErrorRetryBehavior} value.
+ */
+ public function getRetryBehavior(): int
+ {
+ return $this->retryBehavior;
+ }
+
+ public function getErrorType(): ErrorType
+ {
+ return ErrorType::tryFrom($this->type) ?? ErrorType::Unknown;
+ }
+
+ public function getRetryBehaviorEnum(): RetryBehavior
+ {
+ return match ($this->retryBehavior) {
+ NexusHandlerErrorRetryBehavior::NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE => RetryBehavior::Retryable,
+ NexusHandlerErrorRetryBehavior::NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE => RetryBehavior::NonRetryable,
+ default => RetryBehavior::Unspecified,
+ };
+ }
+}
diff --git a/src/Exception/Failure/NexusOperationFailure.php b/src/Exception/Failure/NexusOperationFailure.php
new file mode 100644
index 000000000..afac4faea
--- /dev/null
+++ b/src/Exception/Failure/NexusOperationFailure.php
@@ -0,0 +1,70 @@
+ $endpoint,
+ 'service' => $service,
+ 'operation' => $operation,
+ 'operationToken' => $operationToken,
+ 'scheduledEventId' => $scheduledEventId,
+ ]),
+ $message,
+ $previous,
+ );
+ }
+
+ public function getScheduledEventId(): int
+ {
+ return $this->scheduledEventId;
+ }
+
+ public function getEndpoint(): string
+ {
+ return $this->endpoint;
+ }
+
+ public function getService(): string
+ {
+ return $this->service;
+ }
+
+ public function getOperation(): string
+ {
+ return $this->operation;
+ }
+
+ /**
+ * Async operation token issued by the handler — empty for sync operations.
+ */
+ public function getOperationToken(): string
+ {
+ return $this->operationToken;
+ }
+}
diff --git a/src/Interceptor/NexusOperationInbound/CancelOperationInput.php b/src/Interceptor/NexusOperationInbound/CancelOperationInput.php
new file mode 100644
index 000000000..6b08d9a10
--- /dev/null
+++ b/src/Interceptor/NexusOperationInbound/CancelOperationInput.php
@@ -0,0 +1,40 @@
+operationContext,
+ $cancelDetails ?? $this->cancelDetails,
+ );
+ }
+}
diff --git a/src/Interceptor/NexusOperationInbound/StartOperationInput.php b/src/Interceptor/NexusOperationInbound/StartOperationInput.php
new file mode 100644
index 000000000..0100cf202
--- /dev/null
+++ b/src/Interceptor/NexusOperationInbound/StartOperationInput.php
@@ -0,0 +1,45 @@
+operationContext,
+ $startDetails ?? $this->startDetails,
+ $input === self::UNSET ? $this->input : $input,
+ );
+ }
+}
diff --git a/src/Interceptor/NexusOperationInboundCallsInterceptor.php b/src/Interceptor/NexusOperationInboundCallsInterceptor.php
new file mode 100644
index 000000000..b75509030
--- /dev/null
+++ b/src/Interceptor/NexusOperationInboundCallsInterceptor.php
@@ -0,0 +1,53 @@
+operationContext->headers->get('authorization') !== 'expected-token') {
+ * throw HandlerException::create(ErrorType::Unauthorized, 'Unauthorized');
+ * }
+ *
+ * return $next($input);
+ * }
+ * }
+ * ```
+ *
+ * @see NexusOperationInboundCallsInterceptorTrait
+ */
+interface NexusOperationInboundCallsInterceptor extends Interceptor
+{
+ /**
+ * @param callable(StartOperationInput): OperationStartResult $next
+ */
+ public function startOperation(StartOperationInput $input, callable $next): OperationStartResult;
+
+ /**
+ * @param callable(CancelOperationInput): void $next
+ */
+ public function cancelOperation(CancelOperationInput $input, callable $next): void;
+}
diff --git a/src/Interceptor/NexusOperationOutbound/GetInfoInput.php b/src/Interceptor/NexusOperationOutbound/GetInfoInput.php
new file mode 100644
index 000000000..2337371d6
--- /dev/null
+++ b/src/Interceptor/NexusOperationOutbound/GetInfoInput.php
@@ -0,0 +1,24 @@
+ $nexusHeaders raw-string headers forwarded
+ * on the Nexus operation wire (separate from the Temporal `Header`
+ * that carries payload-typed values).
+ *
+ * @no-named-arguments
+ * @internal Don't use the constructor. Use {@see self::with()} instead.
+ */
+ public function __construct(
+ public readonly string $endpoint,
+ public readonly string $service,
+ public readonly string $operation,
+ public readonly array $args,
+ public readonly NexusOperationOptions $options,
+ public readonly null|Type|string|\ReflectionClass|\ReflectionType $returnType,
+ public readonly array $nexusHeaders = [],
+ ) {
+ if ($endpoint === '') {
+ throw new \InvalidArgumentException('$endpoint must be a non-empty string.');
+ }
+ if ($service === '') {
+ throw new \InvalidArgumentException('$service must be a non-empty string.');
+ }
+ if ($operation === '') {
+ throw new \InvalidArgumentException('$operation must be a non-empty string.');
+ }
+ }
+
+ public function with(
+ ?string $endpoint = null,
+ ?string $service = null,
+ ?string $operation = null,
+ ?array $args = null,
+ ?NexusOperationOptions $options = null,
+ null|Type|string|\ReflectionClass|\ReflectionType $returnType = null,
+ ?array $nexusHeaders = null,
+ ): self {
+ return new self(
+ $endpoint ?? $this->endpoint,
+ $service ?? $this->service,
+ $operation ?? $this->operation,
+ $args ?? $this->args,
+ $options ?? $this->options,
+ $returnType ?? $this->returnType,
+ $nexusHeaders ?? $this->nexusHeaders,
+ );
+ }
+}
diff --git a/src/Interceptor/WorkflowOutboundCallsInterceptor.php b/src/Interceptor/WorkflowOutboundCallsInterceptor.php
index 620f1a4f8..8c72edecc 100644
--- a/src/Interceptor/WorkflowOutboundCallsInterceptor.php
+++ b/src/Interceptor/WorkflowOutboundCallsInterceptor.php
@@ -21,6 +21,7 @@
use Temporal\Interceptor\WorkflowOutboundCalls\ExecuteActivityInput;
use Temporal\Interceptor\WorkflowOutboundCalls\ExecuteChildWorkflowInput;
use Temporal\Interceptor\WorkflowOutboundCalls\ExecuteLocalActivityInput;
+use Temporal\Interceptor\WorkflowOutboundCalls\ExecuteNexusOperationInput;
use Temporal\Interceptor\WorkflowOutboundCalls\GetVersionInput;
use Temporal\Interceptor\WorkflowOutboundCalls\PanicInput;
use Temporal\Interceptor\WorkflowOutboundCalls\SideEffectInput;
@@ -141,4 +142,9 @@ public function await(AwaitInput $input, callable $next): PromiseInterface;
* @param callable(AwaitWithTimeoutInput): PromiseInterface $next
*/
public function awaitWithTimeout(AwaitWithTimeoutInput $input, callable $next): PromiseInterface;
+
+ /**
+ * @param callable(ExecuteNexusOperationInput): PromiseInterface $next
+ */
+ public function executeNexusOperation(ExecuteNexusOperationInput $input, callable $next): PromiseInterface;
}
diff --git a/src/Internal/Client/OnConflictOptions.php b/src/Internal/Client/OnConflictOptions.php
new file mode 100644
index 000000000..25ec82d5a
--- /dev/null
+++ b/src/Internal/Client/OnConflictOptions.php
@@ -0,0 +1,28 @@
+setUrl($callback->url);
+ if ($callback->headers !== []) {
+ $nexus->setHeader($callback->headers);
+ }
+
+ $proto = new Callback();
+ $proto->setNexus($nexus);
+ if ($callback->links !== []) {
+ $proto->setLinks($callback->links);
+ }
+ return $proto;
+ }
+
+ private static function onConflictOptionsToProto(OnConflictOptions $options): OnConflictOptionsProto
+ {
+ $proto = new OnConflictOptionsProto();
+ $proto->setAttachRequestId($options->attachRequestId);
+ $proto->setAttachCompletionCallbacks($options->attachCompletionCallbacks);
+ $proto->setAttachLinks($options->attachLinks);
+ return $proto;
+ }
+
/**
* @param StartWorkflowExecutionRequest|SignalWithStartWorkflowExecutionRequest $request
* use {@see configureExecutionRequest()} to prepare request
@@ -324,7 +353,7 @@ private function configureExecutionRequest(
): StartWorkflowExecutionRequest|SignalWithStartWorkflowExecutionRequest {
$options = $input->options;
- $req->setRequestId(Uuid::v4())
+ $req->setRequestId($options->requestId ?? Uuid::v4())
->setIdentity($this->clientOptions->identity)
->setNamespace($this->clientOptions->namespace)
->setTaskQueue(new TaskQueue(['name' => $options->taskQueue]))
@@ -395,6 +424,21 @@ private function configureExecutionRequest(
if ($req instanceof StartWorkflowExecutionRequest) {
$req->setRequestEagerExecution($options->eagerStart);
+
+ // SignalWithStart's proto has no completion_callbacks / on_conflict_options field — start path only.
+ if ($options->completionCallbacks !== []) {
+ $req->setCompletionCallbacks(
+ \array_map(self::completionCallbackToProto(...), $options->completionCallbacks),
+ );
+ }
+
+ if ($options->onConflictOptions !== null) {
+ $req->setOnConflictOptions(self::onConflictOptionsToProto($options->onConflictOptions));
+ }
+
+ if ($options->links !== []) {
+ $req->setLinks($options->links);
+ }
}
if (!$input->arguments->isEmpty()) {
diff --git a/src/Internal/Declaration/Graph/ClassNode.php b/src/Internal/Declaration/Graph/ClassNode.php
index 996b894bc..2d59fe5a8 100644
--- a/src/Internal/Declaration/Graph/ClassNode.php
+++ b/src/Internal/Declaration/Graph/ClassNode.php
@@ -58,7 +58,7 @@ public function count(): int
* Get a method with all the declared classes.
*
* @param non-empty-string $name
- * @return \Traversable>
+ * @return \Traversable>
* @throws \ReflectionException
*/
public function getMethods(string $name, bool $reverse = true): \Traversable
@@ -190,7 +190,7 @@ private function getInheritance(): array
/**
* @param iterable $classes
- * @return list
+ * @return list
* @throws \ReflectionException
*/
private function boxMethods(iterable $classes, string $name): array
@@ -207,8 +207,8 @@ private function boxMethods(iterable $classes, string $name): array
}
/**
- * @param array $boxed
- * @return \Traversable
+ * @param array $boxed
+ * @return \Traversable
*/
private function unboxMethods(array $boxed): \Traversable
{
diff --git a/src/Internal/Declaration/Instantiator/NexusServiceInstantiator.php b/src/Internal/Declaration/Instantiator/NexusServiceInstantiator.php
new file mode 100644
index 000000000..a801dd75d
--- /dev/null
+++ b/src/Internal/Declaration/Instantiator/NexusServiceInstantiator.php
@@ -0,0 +1,114 @@
+resolveInstance($prototype);
+ $handlers = $this->buildOperationHandlers($prototype, $instance);
+
+ return new NexusServiceInstance($prototype, $handlers);
+ }
+
+ private function resolveInstance(NexusServicePrototype $prototype): object
+ {
+ $factory = $prototype->getFactory();
+ if ($factory !== null) {
+ $instance = $factory();
+ if (!\is_object($instance)) {
+ throw new InvalidArgumentException(\sprintf(
+ 'Nexus service factory for "%s" must return an object, got %s',
+ $prototype->getID(),
+ \get_debug_type($instance),
+ ));
+ }
+ return $instance;
+ }
+
+ $reflection = $prototype->getClass();
+ if (!$reflection->isInstantiable() || $reflection->getConstructor()?->getNumberOfRequiredParameters() > 0) {
+ throw new NexusException(\sprintf(
+ 'Service implementation for "%s" cannot be instantiated without arguments — bind via withInstance() or withFactory()',
+ $prototype->getID(),
+ ));
+ }
+
+ return $reflection->newInstance();
+ }
+
+ /**
+ * @return array
+ */
+ private function buildOperationHandlers(NexusServicePrototype $prototype, object $instance): array
+ {
+ $handlers = [];
+ $reflection = new \ReflectionClass($instance);
+
+ foreach ($prototype->getOperations() as $operation) {
+ try {
+ $startMethod = $reflection->getMethod($operation->methodName);
+ } catch (\ReflectionException $e) {
+ throw new NexusException(\sprintf(
+ 'Service implementation %s is missing method %s() for operation "%s"',
+ $reflection->getName(),
+ $operation->methodName,
+ $operation->name,
+ ), 0, $e);
+ }
+
+ if (!$startMethod->isPublic() || $startMethod->isStatic() || $startMethod->isAbstract()) {
+ throw new NexusException(\sprintf(
+ 'Operation method %s::%s() must be public and non-static',
+ $reflection->getName(),
+ $operation->methodName,
+ ));
+ }
+
+ if (NexusServiceReader::returnsOperationHandler($startMethod)) {
+ $handler = $startMethod->invoke($instance);
+ if (!$handler instanceof OperationHandlerInterface) {
+ throw new NexusException(\sprintf(
+ 'Operation handler factory %s::%s() must return an %s instance, got %s',
+ $reflection->getName(),
+ $operation->methodName,
+ OperationHandlerInterface::class,
+ \get_debug_type($handler),
+ ));
+ }
+ $handlers[$operation->name] = $handler;
+ continue;
+ }
+
+ $handlers[$operation->name] = new MethodOperationHandler(
+ instance: $instance,
+ startMethod: $startMethod,
+ operation: $operation,
+ );
+ }
+
+ return $handlers;
+ }
+}
diff --git a/src/Internal/Declaration/NexusServiceInstance.php b/src/Internal/Declaration/NexusServiceInstance.php
new file mode 100644
index 000000000..532cfc9ec
--- /dev/null
+++ b/src/Internal/Declaration/NexusServiceInstance.php
@@ -0,0 +1,30 @@
+ $operationHandlers Keyed by wire operation name.
+ */
+ public function __construct(
+ public readonly NexusServicePrototype $prototype,
+ public readonly array $operationHandlers,
+ ) {}
+}
diff --git a/src/Internal/Declaration/Prototype/NexusOperationPrototype.php b/src/Internal/Declaration/Prototype/NexusOperationPrototype.php
new file mode 100644
index 000000000..e30aca3eb
--- /dev/null
+++ b/src/Internal/Declaration/Prototype/NexusOperationPrototype.php
@@ -0,0 +1,36 @@
+
+ */
+final class NexusServiceCollection extends ArrayRepository {}
diff --git a/src/Internal/Declaration/Prototype/NexusServicePrototype.php b/src/Internal/Declaration/Prototype/NexusServicePrototype.php
new file mode 100644
index 000000000..458d95627
--- /dev/null
+++ b/src/Internal/Declaration/Prototype/NexusServicePrototype.php
@@ -0,0 +1,69 @@
+ Keyed by wire operation name. */
+ private readonly array $operations;
+
+ private ?\Closure $factory = null;
+
+ /**
+ * @param non-empty-string $name Wire-level service name.
+ * @param array $operations
+ */
+ public function __construct(
+ string $name,
+ array $operations,
+ \ReflectionClass $class,
+ ) {
+ ServiceNameValidator::assert($name);
+ parent::__construct($name, null, $class);
+ $this->operations = $operations;
+ }
+
+ /**
+ * @return array
+ */
+ public function getOperations(): array
+ {
+ return $this->operations;
+ }
+
+ public function getFactory(): ?\Closure
+ {
+ return $this->factory;
+ }
+
+ /**
+ * Bind the given object as the service implementation. Equivalent to
+ * `withFactory(static fn() => $instance)` — same shape as
+ * {@see ActivityPrototype::withInstance()}.
+ */
+ public function withInstance(object $instance): self
+ {
+ return $this->withFactory(static fn(): object => $instance);
+ }
+
+ public function withFactory(\Closure $factory): self
+ {
+ $proto = clone $this;
+ $proto->factory = $factory;
+ return $proto;
+ }
+}
diff --git a/src/Internal/Declaration/Reader/ActivityReader.php b/src/Internal/Declaration/Reader/ActivityReader.php
index c97c22a4c..5b00fed67 100644
--- a/src/Internal/Declaration/Reader/ActivityReader.php
+++ b/src/Internal/Declaration/Reader/ActivityReader.php
@@ -96,7 +96,7 @@ private function getMethodGroups(ClassNode $graph, \ReflectionMethod $root): arr
// Each group of methods means one level of hierarchy in the
// inheritance graph.
//
- /** @var ClassNode $ctx */
+ /** @var \Traversable $group */
foreach ($group as $ctx => $method) {
/** @var MethodRetry $retry */
$retry = $this->reader->firstFunctionMetadata($method, MethodRetry::class);
diff --git a/src/Internal/Declaration/Reader/NexusServiceReader.php b/src/Internal/Declaration/Reader/NexusServiceReader.php
new file mode 100644
index 000000000..9aee1474a
--- /dev/null
+++ b/src/Internal/Declaration/Reader/NexusServiceReader.php
@@ -0,0 +1,299 @@
+
+ */
+class NexusServiceReader extends Reader
+{
+ /**
+ * @internal
+ */
+ public static function returnsOperationHandler(\ReflectionMethod $method): bool
+ {
+ $returnType = $method->getReturnType();
+
+ return $returnType instanceof \ReflectionNamedType
+ && !$returnType->allowsNull()
+ && !$returnType->isBuiltin()
+ && \is_a($returnType->getName(), OperationHandlerInterface::class, true);
+ }
+
+ /**
+ * @param class-string $class
+ */
+ public function fromClass(string $class): NexusServicePrototype
+ {
+ $reflection = new \ReflectionClass($class);
+ $graph = new ClassNode($reflection);
+
+ $serviceNodes = $this->serviceNodes($graph);
+ $contract = $this->resolveContract($reflection, $serviceNodes);
+
+ $service = $this->reader->firstClassMetadata($contract, Service::class);
+ // resolveContract guarantees a Service attribute is present.
+ \assert($service !== null);
+ $name = $service->name !== '' ? $service->name : $contract->getShortName();
+
+ $this->assertServiceNamesMatch($serviceNodes, $contract, $name);
+
+ $operations = $this->collectOperations($graph);
+
+ return new NexusServicePrototype($name, $operations, $reflection);
+ }
+
+ /**
+ * Locate the single `#[Service]`-bearing type for the hierarchy; the root wins if annotated.
+ *
+ * @param \ReflectionClass