feat(lua): workflow history replayer module

boot/components/runtime/lua/all.go

@@ -33,6 +33,7 @@ func All() []boot.Component {
 		Payload(),
 		Queue(),
 		Registry(),
+		Replayer(),
 		Security(),
 		SQL(),
 		Store(),

boot/components/runtime/lua/constants.go

@@ -27,6 +27,7 @@ const (
 	MetricsName      = "lua.metrics"
 	JSONName         = "lua.json"
 	PayloadName      = "lua.payload"
+	ReplayerName     = "lua.replayer"
 	RegistryName     = "lua.registry"
 	SecurityName     = "lua.security"
 	SQLName          = "lua.sql"

boot/components/runtime/lua/replayer.go

@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: MPL-2.0
+
+package lua
+
+import (
+	"context"
+
+	"github.com/wippyai/runtime/api/boot"
+	"github.com/wippyai/runtime/runtime/lua/modules/replayer"
+)
+
+func Replayer() boot.Component {
+	return boot.New(boot.P{
+		Name:      ReplayerName,
+		DependsOn: []boot.Name{EngineName},
+		Load: func(ctx context.Context) (context.Context, error) {
+			cm := GetCodeManager(ctx)
+			if cm == nil {
+				return ctx, nil
+			}
+
+			if err := AddModules(ctx, cm,
+				replayer.Module,
+			); err != nil {
+				return ctx, err
+			}
+
+			return ctx, nil
+		},
+	})
+}

runtime/lua/modules/replayer/module.go

@@ -0,0 +1,102 @@
+// SPDX-License-Identifier: MPL-2.0
+
+// Package replayer exposes Temporal's workflow history replayer to Lua.
+package replayer
+
+import (
+	lua "github.com/wippyai/go-lua"
+	"github.com/wippyai/runtime/api/registry"
+	luaapi "github.com/wippyai/runtime/api/runtime/lua"
+	temporalapi "github.com/wippyai/runtime/api/service/temporal"
+	tpropagator "github.com/wippyai/runtime/service/temporal/propagator"
+	tworkflow "github.com/wippyai/runtime/service/temporal/workflow"
+	"go.temporal.io/sdk/worker"
+	sdkworkflow "go.temporal.io/sdk/workflow"
+)
+
+// Module is the replayer Lua module (ClassIO: usable from tests, not workflow bodies).
+var Module = &luaapi.ModuleDef{
+	Name:        "replayer",
+	Description: "Temporal workflow history replay for determinism tests",
+	Class:       []string{luaapi.ClassIO, luaapi.ClassNondeterministic},
+	Build:       buildModule,
+	Types:       ModuleTypes,
+}
+
+func buildModule() (*lua.LTable, []luaapi.YieldType) {
+	mod := lua.CreateTable(0, 1)
+	mod.RawSetString("replay_json_file", lua.LGoFunc(replayJSONFile))
+	mod.Immutable = true
+	return mod, nil
+}
+
+func invalidError(l *lua.LState, msg string) int {
+	err := lua.NewLuaError(l, msg).WithKind(lua.Invalid).WithRetryable(false)
+	l.Push(lua.LNil)
+	l.Push(err)
+	return 2
+}
+
+func internalError(l *lua.LState, goErr error, context string) int {
+	err := lua.WrapErrorWithLua(l, goErr, context).WithKind(lua.Internal).WithRetryable(false)
+	l.Push(lua.LNil)
+	l.Push(err)
+	return 2
+}
+
+// replayJSONFile(workflow_id, history_json_path[, workflow_type_name]) -> (true) | (nil, err);
+// nil err == deterministic replay. workflow_type_name overrides the registered type name
+// when the workflow uses a custom meta name (defaults to workflow_id).
+func replayJSONFile(l *lua.LState) int {
+	workflowID := l.CheckString(1)
+	path := l.CheckString(2)
+	if workflowID == "" {
+		return invalidError(l, "workflow id required")
+	}
+	if path == "" {
+		return invalidError(l, "history file path required")
+	}
+
+	regID := registry.ParseID(workflowID)
+	if regID.NS == "" || regID.Name == "" {
+		return invalidError(l, "workflow id must be in 'namespace:name' form")
+	}
+
+	typeName := workflowID
+	if tn := l.OptString(3, ""); tn != "" {
+		typeName = tn
+	}
+
+	ctx := l.Context()
+
+	dcReg := temporalapi.GetDataConverterRegistry(ctx)
+	if dcReg == nil {
+		return invalidError(l, "temporal data converter registry not available in context")
+	}
+	dc := dcReg.Build()
+
+	opts := worker.WorkflowReplayerOptions{
+		DataConverter:      dc,
+		ContextPropagators: []sdkworkflow.ContextPropagator{tpropagator.New(dc)},
+	}
+	if wreg := temporalapi.GetWorkerInterceptorRegistry(ctx); wreg != nil {
+		opts.Interceptors = wreg.GetAll()
+	}
+
+	r, err := worker.NewWorkflowReplayerWithOptions(opts)
+	if err != nil {
+		return internalError(l, err, "create workflow replayer")
+	}
+
+	// Register the runtime's dynamic definition (SDK accepts it as a WorkflowDefinitionFactory).
+	factory := (&tworkflow.DefinitionFactory{ID: regID}).WithContext(ctx)
+	r.RegisterWorkflowWithOptions(factory, sdkworkflow.RegisterOptions{Name: typeName})
+
+	if err := r.ReplayWorkflowHistoryFromJSONFile(nil, path); err != nil {
+		return internalError(l, err, "replay")
+	}
+
+	l.Push(lua.LBool(true))
+	l.Push(lua.LNil)
+	return 2
+}

runtime/lua/modules/replayer/module_test.go

@@ -0,0 +1,65 @@
+// SPDX-License-Identifier: MPL-2.0
+
+package replayer
+
+import (
+	"testing"
+
+	lua "github.com/wippyai/go-lua"
+)
+
+func newBound(t *testing.T) *lua.LState {
+	t.Helper()
+	l := lua.NewState()
+	tbl, _ := Module.Build()
+	l.SetGlobal(Module.Name, tbl)
+	return l
+}
+
+func TestBind(t *testing.T) {
+	l := newBound(t)
+	defer l.Close()
+
+	mod := l.GetGlobal("replayer")
+	if mod.Type() != lua.LTTable {
+		t.Fatal("replayer module not registered")
+	}
+	if mod.(*lua.LTable).RawGetString("replay_json_file").Type() != lua.LTFunction {
+		t.Error("replay_json_file function not registered")
+	}
+}
+
+// Argument validation runs before any context/Temporal access, so this needs no runtime.
+func TestReplayBadArgs(t *testing.T) {
+	l := newBound(t)
+	defer l.Close()
+
+	err := l.DoString(`
+		local _, err = replayer.replay_json_file("", "history.json")
+		if err == nil then error("expected error for empty workflow id") end
+
+		_, err = replayer.replay_json_file("no-colon-here", "history.json")
+		if err == nil then error("expected error for non ns:name id") end
+
+		_, err = replayer.replay_json_file("app:wf", "")
+		if err == nil then error("expected error for empty path") end
+	`)
+	if err != nil {
+		t.Errorf("bad-args test failed: %v", err)
+	}
+}
+
+// Well-formed args with no Temporal in context must fail gracefully, not panic.
+func TestReplayNoTemporalContext(t *testing.T) {
+	l := newBound(t)
+	defer l.Close()
+
+	err := l.DoString(`
+		local ok, err = replayer.replay_json_file("app.letter:scheduled_delivery_workflow", "history.json")
+		if ok ~= nil then error("expected nil result without temporal context") end
+		if err == nil then error("expected error without temporal context") end
+	`)
+	if err != nil {
+		t.Errorf("no-temporal-context test failed: %v", err)
+	}
+}

runtime/lua/modules/replayer/types.go

@@ -0,0 +1,23 @@
+// SPDX-License-Identifier: MPL-2.0
+
+package replayer
+
+import (
+	"github.com/wippyai/go-lua/types/io"
+	"github.com/wippyai/go-lua/types/typ"
+)
+
+// ModuleTypes returns the type manifest for the replayer module.
+func ModuleTypes() *io.Manifest {
+	m := io.NewManifest("replayer")
+
+	moduleType := typ.NewInterface("replayer", []typ.Method{
+		{Name: "replay_json_file", Type: typ.Func().
+			Param("workflow_id", typ.Any).
+			Param("history_json_path", typ.Any).
+			Returns(typ.Any, typ.NewOptional(typ.LuaError)).Build()},
+	})
+
+	m.SetExport(moduleType)
+	return m
+}

service/temporal/workflow/replay_integration_test.go

@@ -0,0 +1,68 @@
+// SPDX-License-Identifier: MPL-2.0
+
+package workflow_test
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/wippyai/runtime/api/registry"
+	"github.com/wippyai/runtime/service/temporal/dataconverter"
+	"github.com/wippyai/runtime/service/temporal/workflow"
+	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/api/temporalproto"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+	sdkworkflow "go.temporal.io/sdk/workflow"
+)
+
+// TestReplay_Integration runs a Lua workflow, then replays its history against the
+// same DefinitionFactory. A nil error proves deterministic replay.
+func TestReplay_Integration(t *testing.T) {
+	wfID := registry.NewID("test.workflow", "replay-hello")
+	f := newWorkflowTestFixture(t, workflowTestOpts{
+		workflowID: wfID,
+		source:     helloWorkflowSource,
+		taskQueue:  "test-replay-queue",
+	})
+	defer f.cleanup()
+
+	run := f.startWorkflow(wfID.String(), map[string]any{"name": "Replay"})
+	var result map[string]any
+	require.NoError(t, run.Get(f.ctx, &result))
+
+	hist := fetchHistory(t, f.ctx, f.temporalClient, run.GetID(), run.GetRunID())
+	require.NotEmpty(t, hist.Events)
+
+	dc := dataconverter.NewDataConverter(newTestTranscoder())
+	replayer, err := worker.NewWorkflowReplayerWithOptions(worker.WorkflowReplayerOptions{DataConverter: dc})
+	require.NoError(t, err)
+
+	factory := (&workflow.DefinitionFactory{ID: wfID}).WithContext(f.ctx)
+	replayer.RegisterWorkflowWithOptions(factory, sdkworkflow.RegisterOptions{Name: wfID.String()})
+
+	require.NoError(t, replayer.ReplayWorkflowHistory(nil, hist))
+
+	// Cover the module's exact path: history written to a JSON file, replayed with a nil logger.
+	data, err := temporalproto.CustomJSONMarshalOptions{}.Marshal(hist)
+	require.NoError(t, err)
+	histFile := filepath.Join(t.TempDir(), "history.json")
+	require.NoError(t, os.WriteFile(histFile, data, 0o600))
+	require.NoError(t, replayer.ReplayWorkflowHistoryFromJSONFile(nil, histFile))
+}
+
+func fetchHistory(t *testing.T, ctx context.Context, c client.Client, wfID, runID string) *historypb.History {
+	t.Helper()
+	it := c.GetWorkflowHistory(ctx, wfID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+	var events []*historypb.HistoryEvent
+	for it.HasNext() {
+		ev, err := it.Next()
+		require.NoError(t, err)
+		events = append(events, ev)
+	}
+	return &historypb.History{Events: events}
+}