worker: expose activity heartbeat recording

Core\Worker::recordActivityHeartbeat(bytes) hands a serialized coresdk
ActivityHeartbeat to the core. The bridge call is synchronous (in-memory
record; throttling and the server RPC run on the core's own threads), so
the method never parks the coroutine and needs no cross-thread call slot.
The finalized guard covers it: the bridge unwraps None after finalize.

Also fixes the stale "activity-only worker" comment in temporal_core.h
(doc drift noted in the 2026-06-11 review).

Author: EdmondDantes

CHANGELOG.md

@@ -36,6 +36,11 @@ transport through a `ServiceClientInterface` adapter — see the
   enabled pollers to wind down.
 - The connection now defaults its identity to `<pid>@<hostname>` when none is
   given (required by workers).
+- `Core\Worker::recordActivityHeartbeat(bytes)` — records an activity heartbeat
+  (serialized coresdk ActivityHeartbeat). Synchronous and never parks the
+  coroutine: the core stores the heartbeat in memory and throttles/sends it to
+  the server on its own threads. A pending cancellation is delivered separately,
+  as a cancel-variant task from `pollActivityTask()`.
 - The worker now caches workflow runs (sticky execution) so a fired timer or
   resolved activity resumes the live instance instead of replaying from scratch.
 

README.md

@@ -49,6 +49,9 @@ Just the transport seam:
 
 - `TrueAsync\Temporal\Core\Connection` — `connect` plus an async
   `rpcCall(service, method, requestBytes): responseBytes`.
+- `TrueAsync\Temporal\Core\Worker` — the worker transport: poll/complete for
+  activity tasks and workflow activations, activity heartbeat recording, and
+  the shutdown lifecycle.
 - `TrueAsync\Temporal\{TemporalException, ConnectionException, ServiceException}`.
 
 Everything user-facing (workflow client, options, data converter, the generated

temporal.stub.php

@@ -82,6 +82,15 @@ public function pollActivityTask(): ?string {}
     /** Report an activity task completion (serialized coresdk completion). */
     public function completeActivityTask(string $completion): void {}
 
+    /**
+     * Record an activity heartbeat (serialized coresdk ActivityHeartbeat).
+     * Synchronous — never parks the coroutine: the core stores the heartbeat in
+     * memory and throttles/sends it to the server on its own threads. A pending
+     * cancellation is delivered separately, as a cancel-variant task from
+     * pollActivityTask().
+     */
+    public function recordActivityHeartbeat(string $heartbeat): void {}
+
     /**
      * Poll for the next workflow activation. Parks the coroutine until an
      * activation is ready; returns the serialized coresdk WorkflowActivation, or

temporal_arginfo.h

@@ -1,5 +1,5 @@
 /* This is a generated file, edit temporal.stub.php instead.
- * Stub hash: e3c06eb5024b0783e98d6eb0477faa44017b1e0c */
+ * Stub hash: 9f5736251d4657a08384ad984b98be8a08100cb8 */
 
 ZEND_BEGIN_ARG_INFO_EX(arginfo_class_TrueAsync_Temporal_Core_Connection___construct, 0, 0, 1)
 	ZEND_ARG_TYPE_INFO(0, address, IS_STRING, 0)
@@ -34,6 +34,10 @@ ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_Temporal_Core_Wo
 	ZEND_ARG_TYPE_INFO(0, completion, IS_STRING, 0)
 ZEND_END_ARG_INFO()
 
+ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(arginfo_class_TrueAsync_Temporal_Core_Worker_recordActivityHeartbeat, 0, 1, IS_VOID, 0)
+	ZEND_ARG_TYPE_INFO(0, heartbeat, IS_STRING, 0)
+ZEND_END_ARG_INFO()
+
 #define arginfo_class_TrueAsync_Temporal_Core_Worker_pollWorkflowActivation arginfo_class_TrueAsync_Temporal_Core_Worker_pollActivityTask
 
 #define arginfo_class_TrueAsync_Temporal_Core_Worker_completeWorkflowActivation arginfo_class_TrueAsync_Temporal_Core_Worker_completeActivityTask
@@ -48,6 +52,7 @@ ZEND_METHOD(TrueAsync_Temporal_Core_Connection, rpcCall);
 ZEND_METHOD(TrueAsync_Temporal_Core_Worker, __construct);
 ZEND_METHOD(TrueAsync_Temporal_Core_Worker, pollActivityTask);
 ZEND_METHOD(TrueAsync_Temporal_Core_Worker, completeActivityTask);
+ZEND_METHOD(TrueAsync_Temporal_Core_Worker, recordActivityHeartbeat);
 ZEND_METHOD(TrueAsync_Temporal_Core_Worker, pollWorkflowActivation);
 ZEND_METHOD(TrueAsync_Temporal_Core_Worker, completeWorkflowActivation);
 ZEND_METHOD(TrueAsync_Temporal_Core_Worker, initiateShutdown);
@@ -63,6 +68,7 @@ static const zend_function_entry class_TrueAsync_Temporal_Core_Worker_methods[]
 	ZEND_ME(TrueAsync_Temporal_Core_Worker, __construct, arginfo_class_TrueAsync_Temporal_Core_Worker___construct, ZEND_ACC_PUBLIC)
 	ZEND_ME(TrueAsync_Temporal_Core_Worker, pollActivityTask, arginfo_class_TrueAsync_Temporal_Core_Worker_pollActivityTask, ZEND_ACC_PUBLIC)
 	ZEND_ME(TrueAsync_Temporal_Core_Worker, completeActivityTask, arginfo_class_TrueAsync_Temporal_Core_Worker_completeActivityTask, ZEND_ACC_PUBLIC)
+	ZEND_ME(TrueAsync_Temporal_Core_Worker, recordActivityHeartbeat, arginfo_class_TrueAsync_Temporal_Core_Worker_recordActivityHeartbeat, ZEND_ACC_PUBLIC)
 	ZEND_ME(TrueAsync_Temporal_Core_Worker, pollWorkflowActivation, arginfo_class_TrueAsync_Temporal_Core_Worker_pollWorkflowActivation, ZEND_ACC_PUBLIC)
 	ZEND_ME(TrueAsync_Temporal_Core_Worker, completeWorkflowActivation, arginfo_class_TrueAsync_Temporal_Core_Worker_completeWorkflowActivation, ZEND_ACC_PUBLIC)
 	ZEND_ME(TrueAsync_Temporal_Core_Worker, initiateShutdown, arginfo_class_TrueAsync_Temporal_Core_Worker_initiateShutdown, ZEND_ACC_PUBLIC)

temporal_client.c

@@ -914,6 +914,34 @@ PHP_METHOD(TrueAsync_Temporal_Core_Worker, completeWorkflowActivation)
 	}
 }
 
+PHP_METHOD(TrueAsync_Temporal_Core_Worker, recordActivityHeartbeat)
+{
+	zend_string *heartbeat;
+
+	ZEND_PARSE_PARAMETERS_START(1, 1)
+		Z_PARAM_STR(heartbeat)
+	ZEND_PARSE_PARAMETERS_END();
+
+	temporal_worker_obj *self = temporal_worker_from_obj(Z_OBJ_P(ZEND_THIS));
+
+	/* No coroutine requirement: recording is synchronous in the core (in-memory
+	 * store; throttling and the server RPC run on the core's own threads). */
+	if (self->worker == NULL || self->finalized) {
+		zend_throw_error(NULL, "recordActivityHeartbeat must be called on a live worker");
+		RETURN_THROWS();
+	}
+
+	char *fail = tphp_worker_record_heartbeat(((tphp_handle_t *) self->worker)->box,
+	                                          (const uint8_t *) ZSTR_VAL(heartbeat),
+	                                          ZSTR_LEN(heartbeat));
+
+	if (fail != NULL) {
+		zend_throw_exception_ex(temporal_ce_service_exception, 0, "%s", fail);
+		free(fail);
+		RETURN_THROWS();
+	}
+}
+
 PHP_METHOD(TrueAsync_Temporal_Core_Worker, initiateShutdown)
 {
 	ZEND_PARSE_PARAMETERS_NONE();

temporal_core.c

@@ -533,3 +533,16 @@ void tphp_worker_finalize_shutdown(void *worker, void *user_data, tphp_worker_do
 
 	temporal_core_worker_finalize_shutdown((TemporalCoreWorker *) worker, c, tphp_on_worker_done);
 }
+
+char *tphp_worker_record_heartbeat(void *worker, const uint8_t *heartbeat, size_t len)
+{
+	TemporalCoreByteArrayRef ref;
+	ref.data = heartbeat;
+	ref.size = len;
+
+	const TemporalCoreByteArray *fail =
+		temporal_core_worker_record_activity_heartbeat((TemporalCoreWorker *) worker, ref);
+
+	size_t fail_len = 0;
+	return tphp_copy_str(tphp_g_runtime, fail, &fail_len);
+}

temporal_core.h

@@ -105,9 +105,10 @@ void tphp_response_free(void *response_owner);
 
 /* --- Worker ------------------------------------------------------------ */
 
-/* Create an activity-only worker on a task queue (fixed-size activity slots =
- * max_concurrent_activities). Returns an opaque handle, or NULL with *err_out
- * (heap, caller frees). worker_new is synchronous in the core. */
+/* Create a worker on a task queue handling both workflow and activity tasks
+ * (fixed-size activity slots = max_concurrent_activities). Returns an opaque
+ * handle, or NULL with *err_out (heap, caller frees). worker_new is synchronous
+ * in the core. */
 void *tphp_worker_new(void *connection, const char *ns, const char *task_queue,
                       uint32_t max_concurrent_activities, char **err_out);
 void  tphp_worker_free(void *worker);
@@ -134,4 +135,10 @@ void tphp_worker_complete_workflow(void *worker, const uint8_t *completion, size
                                    void *user_data, tphp_worker_done_cb done);
 void tphp_worker_finalize_shutdown(void *worker, void *user_data, tphp_worker_done_cb done);
 
+/* Record an activity heartbeat (serialized coresdk ActivityHeartbeat).
+ * Synchronous in the core: it only stores the heartbeat in memory; throttling
+ * and the server RPC happen later on the core's own threads. Returns NULL on
+ * success, or a heap error message (decode failure) the caller frees. */
+char *tphp_worker_record_heartbeat(void *worker, const uint8_t *heartbeat, size_t len);
+
 #endif /* PHP_TEMPORAL_CORE_H */

tests/live/017-record-heartbeat.phpt

@@ -0,0 +1,53 @@
+--TEST--
+Core\Worker recordActivityHeartbeat: accepts a heartbeat, rejects garbage and a finalized worker
+--EXTENSIONS--
+temporal
+--SKIPIF--
+<?php
+require __DIR__ . '/../inc/temporal.inc';
+temporal_skip_if_no_server();
+?>
+--FILE--
+<?php
+require __DIR__ . '/../inc/temporal.inc';
+
+use TrueAsync\Temporal\Core\Connection;
+use TrueAsync\Temporal\Core\Worker;
+use TrueAsync\Temporal\ServiceException;
+use function Async\spawn;
+use function Async\await;
+
+await(spawn(function () {
+    $conn = new Connection(temporal_test_address());
+    $worker = new Worker($conn, 'hb-' . bin2hex(random_bytes(3)), temporal_test_namespace(), 4);
+
+    // A hand-encoded coresdk ActivityHeartbeat: field 1 (task_token) = "abc".
+    // The token matches no running activity; the core records and ignores it.
+    $worker->recordActivityHeartbeat("\x0a\x03abc");
+    var_dump('recorded');
+
+    // Bytes that are not a valid protobuf message: the core reports a decode
+    // failure, surfaced as a ServiceException.
+    try {
+        $worker->recordActivityHeartbeat("\xff");
+    } catch (ServiceException $e) {
+        var_dump(str_contains($e->getMessage(), 'decode'));
+    }
+
+    $worker->initiateShutdown();
+    $worker->pollWorkflowActivation();
+    $worker->pollActivityTask();
+    $worker->finalizeShutdown();
+
+    // The finalized guard covers heartbeats too (the core would abort otherwise).
+    try {
+        $worker->recordActivityHeartbeat("\x0a\x03abc");
+    } catch (Error $e) {
+        var_dump($e->getMessage());
+    }
+}));
+?>
+--EXPECT--
+string(8) "recorded"
+bool(true)
+string(55) "recordActivityHeartbeat must be called on a live worker"