a3s_flow/
runtime.rs

1use async_trait::async_trait;
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4#[cfg(feature = "native-ts")]
5use sha2::{Digest, Sha256};
6#[cfg(feature = "native-ts")]
7use std::path::Path;
8use std::path::PathBuf;
9use std::time::Duration;
10
11use crate::context::WorkflowContext;
12use crate::error::{FlowError, Result};
13#[cfg(feature = "native-ts")]
14use crate::model::RuntimeKind;
15use crate::model::{FlowEventEnvelope, JsonValue, RuntimeCommand, WorkflowSpec};
16#[cfg(feature = "native-ts")]
17use crate::protocol::{
18    NativeRuntimeKind, NativeRuntimeRequest, NativeRuntimeResponse, NATIVE_RUNTIME_PROTOCOL,
19};
20#[cfg(feature = "native-ts")]
21use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
22#[cfg(feature = "native-ts")]
23use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
24
25#[cfg(feature = "native-ts")]
26mod native_ts;
27
28#[cfg(feature = "native-ts")]
29use native_ts::{
30    artifact_binary_path, verified_dependency_graph, ArtifactCache, ArtifactCacheState,
31    CompilerIdentityCache, SourceSnapshot, TemporaryCacheEntryGuard, VerifiedDependencyGraph,
32    MAX_DEPENDENCY_MANIFEST_BYTES,
33};
34
35/// Workflow replay request passed to a runtime implementation.
36#[derive(Debug, Clone, Serialize, Deserialize)]
37#[non_exhaustive]
38pub struct WorkflowInvocation {
39    /// Durable identifier of the workflow run being replayed.
40    pub run_id: String,
41    /// Workflow definition recorded when the run was created.
42    pub spec: WorkflowSpec,
43    /// Original workflow input.
44    pub input: JsonValue,
45    /// Complete persisted event history in sequence order.
46    pub history: Vec<FlowEventEnvelope>,
47}
48
49impl WorkflowInvocation {
50    /// Create a workflow invocation from its complete durable replay input.
51    pub fn new(
52        run_id: impl Into<String>,
53        spec: WorkflowSpec,
54        input: JsonValue,
55        history: Vec<FlowEventEnvelope>,
56    ) -> Self {
57        Self {
58            run_id: run_id.into(),
59            spec,
60            input,
61            history,
62        }
63    }
64
65    /// Build a deterministic helper view over this workflow invocation.
66    pub fn context(&self) -> WorkflowContext<'_> {
67        WorkflowContext::new(self)
68    }
69
70    /// Decode the workflow input into a host-defined serde type.
71    pub fn input_as<T>(&self) -> Result<T>
72    where
73        T: DeserializeOwned,
74    {
75        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
76    }
77}
78
79/// Step execution request passed to a runtime implementation.
80#[derive(Debug, Clone, Serialize, Deserialize)]
81#[non_exhaustive]
82pub struct StepInvocation {
83    /// Durable identifier of the workflow run that scheduled the step.
84    pub run_id: String,
85    /// Replay-stable identifier of this step invocation.
86    pub step_id: String,
87    /// Host-defined step handler name.
88    pub step_name: String,
89    /// Input supplied by the workflow command.
90    pub input: JsonValue,
91    /// Complete persisted workflow history in sequence order.
92    pub history: Vec<FlowEventEnvelope>,
93}
94
95impl StepInvocation {
96    /// Create a step invocation from its complete durable execution input.
97    pub fn new(
98        run_id: impl Into<String>,
99        step_id: impl Into<String>,
100        step_name: impl Into<String>,
101        input: JsonValue,
102        history: Vec<FlowEventEnvelope>,
103    ) -> Self {
104        Self {
105            run_id: run_id.into(),
106            step_id: step_id.into(),
107            step_name: step_name.into(),
108            input,
109            history,
110        }
111    }
112
113    /// Decode the step input into a host-defined serde type.
114    pub fn input_as<T>(&self) -> Result<T>
115    where
116        T: DeserializeOwned,
117    {
118        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
119    }
120}
121
122/// Runtime boundary for workflow code and side-effecting steps.
123#[async_trait]
124pub trait FlowRuntime: Send + Sync {
125    /// Replay the deterministic workflow function and return the next command.
126    async fn run_workflow(&self, invocation: WorkflowInvocation) -> Result<RuntimeCommand>;
127
128    /// Execute one side-effecting step. The engine persists success/failure.
129    async fn run_step(&self, invocation: StepInvocation) -> Result<JsonValue>;
130}
131
132/// Source-identity policy for native TypeScript artifacts.
133#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
134#[non_exhaustive]
135pub enum NativeTsDependencyMode {
136    /// Preserve the original entrypoint-plus-deployment-version identity.
137    #[default]
138    EntrypointOnly,
139    /// Ask the compiler for its complete source graph and verify every file.
140    CompilerManifest,
141}
142
143/// Configuration for the native TypeScript runtime adapter.
144#[derive(Debug, Clone)]
145#[non_exhaustive]
146pub struct NativeTsRuntimeConfig {
147    /// Compiler executable. Bare names use `PATH`; relative paths with a
148    /// directory component are resolved against the host process directory.
149    pub compiler_binary: PathBuf,
150    /// Artifact cache directory. Relative paths are resolved against the host
151    /// process directory before the compiler changes its working directory.
152    pub cache_dir: PathBuf,
153    /// Runtime working directory. Relative paths are resolved against the host
154    /// process directory, and workflow entrypoints are resolved from it.
155    pub working_dir: PathBuf,
156}
157
158impl NativeTsRuntimeConfig {
159    /// Create a native TypeScript configuration from explicit host paths.
160    pub fn new(
161        compiler_binary: impl Into<PathBuf>,
162        cache_dir: impl Into<PathBuf>,
163        working_dir: impl Into<PathBuf>,
164    ) -> Self {
165        Self {
166            compiler_binary: compiler_binary.into(),
167            cache_dir: cache_dir.into(),
168            working_dir: working_dir.into(),
169        }
170    }
171}
172
173impl Default for NativeTsRuntimeConfig {
174    fn default() -> Self {
175        Self {
176            compiler_binary: PathBuf::from("a3s-flow-native-compiler"),
177            cache_dir: PathBuf::from(".a3s/flow/native-ts"),
178            working_dir: PathBuf::from("."),
179        }
180    }
181}
182
183/// Runtime that compiles TypeScript to a native executable and speaks JSON over
184/// stdin/stdout with that executable.
185#[derive(Debug, Clone)]
186pub struct NativeTsRuntime {
187    config: NativeTsRuntimeConfig,
188    max_stdout_bytes: usize,
189    max_stderr_bytes: usize,
190    compile_timeout: Option<Duration>,
191    invocation_timeout: Option<Duration>,
192    dependency_mode: NativeTsDependencyMode,
193    #[cfg(feature = "native-ts")]
194    compiler_identity_cache: CompilerIdentityCache,
195    #[cfg(feature = "native-ts")]
196    artifact_cache: ArtifactCache,
197}
198
199#[cfg(feature = "native-ts")]
200#[derive(Debug, Clone)]
201struct NativeArtifact {
202    compiler_binary: PathBuf,
203    working_dir: PathBuf,
204    entrypoint: PathBuf,
205    cache_entry: PathBuf,
206    cache_key: String,
207    binary: PathBuf,
208    source_hash: String,
209    source_snapshot: SourceSnapshot,
210    dependency_compiler_identity: Option<String>,
211}
212
213#[cfg(feature = "native-ts")]
214struct NativeProcessOutput {
215    status: std::process::ExitStatus,
216    stdout: Vec<u8>,
217    stderr: Vec<u8>,
218}
219
220#[cfg(feature = "native-ts")]
221#[derive(Debug)]
222enum NativeProcessOutputError {
223    Io(std::io::Error),
224    LimitExceeded { stream: &'static str, limit: usize },
225}
226
227/// Result of validating and compiling a native TypeScript workflow source.
228#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
229#[non_exhaustive]
230pub struct NativeTsRuntimePreflight {
231    /// Resolved workflow source entrypoint used by the compiler.
232    pub entrypoint: PathBuf,
233    /// Resolved native artifact path that will be invoked by the runtime.
234    pub artifact: PathBuf,
235    /// Stable hash of the workflow source and runtime identity fields.
236    pub source_hash: String,
237    /// True when the existing artifact cache entry was reused.
238    pub cache_hit: bool,
239}
240
241impl NativeTsRuntime {
242    /// Default maximum bytes retained from a compiler or runtime stdout pipe.
243    pub const DEFAULT_MAX_STDOUT_BYTES: usize = 8 * 1024 * 1024;
244
245    /// Default maximum bytes retained from a compiler or runtime stderr pipe.
246    pub const DEFAULT_MAX_STDERR_BYTES: usize = 256 * 1024;
247
248    /// Create a native TypeScript runtime with default limits and cache policy.
249    pub fn new(config: NativeTsRuntimeConfig) -> Self {
250        Self {
251            config,
252            max_stdout_bytes: Self::DEFAULT_MAX_STDOUT_BYTES,
253            max_stderr_bytes: Self::DEFAULT_MAX_STDERR_BYTES,
254            compile_timeout: None,
255            invocation_timeout: None,
256            dependency_mode: NativeTsDependencyMode::default(),
257            #[cfg(feature = "native-ts")]
258            compiler_identity_cache: CompilerIdentityCache::default(),
259            #[cfg(feature = "native-ts")]
260            artifact_cache: ArtifactCache::default(),
261        }
262    }
263
264    /// Return the compiler, cache, and working-directory configuration.
265    pub fn config(&self) -> &NativeTsRuntimeConfig {
266        &self.config
267    }
268
269    /// Override the independent byte limits for each compiler and runtime
270    /// stdout/stderr pipe.
271    ///
272    /// Exceeding either limit terminates the direct child process and returns a
273    /// runtime error. A zero limit allows an empty pipe but rejects its first
274    /// byte of output.
275    pub fn with_output_limits(mut self, max_stdout_bytes: usize, max_stderr_bytes: usize) -> Self {
276        self.max_stdout_bytes = max_stdout_bytes;
277        self.max_stderr_bytes = max_stderr_bytes;
278        self
279    }
280
281    /// Return the configured byte limit for each stdout pipe.
282    pub fn max_stdout_bytes(&self) -> usize {
283        self.max_stdout_bytes
284    }
285
286    /// Return the configured byte limit for each stderr pipe.
287    pub fn max_stderr_bytes(&self) -> usize {
288        self.max_stderr_bytes
289    }
290
291    /// Set the maximum duration of each cold compiler process.
292    ///
293    /// Cache hits do not start a compiler and therefore do not consume this
294    /// timeout. By default, compilation has no runtime-owned timeout and
295    /// remains bounded only by caller cancellation or an outer host timeout.
296    pub fn with_compile_timeout(mut self, timeout: Duration) -> Self {
297        self.compile_timeout = Some(timeout);
298        self
299    }
300
301    /// Set the maximum duration of each workflow or step artifact invocation.
302    ///
303    /// The timeout covers writing the complete request to stdin, reading both
304    /// output pipes, and waiting for process exit. By default, invocation has
305    /// no runtime-owned timeout and remains bounded only by caller cancellation
306    /// or an outer host timeout.
307    pub fn with_invocation_timeout(mut self, timeout: Duration) -> Self {
308        self.invocation_timeout = Some(timeout);
309        self
310    }
311
312    /// Return the configured cold-compilation timeout, if any.
313    pub fn compile_timeout(&self) -> Option<Duration> {
314        self.compile_timeout
315    }
316
317    /// Return the configured workflow and step invocation timeout, if any.
318    pub fn invocation_timeout(&self) -> Option<Duration> {
319        self.invocation_timeout
320    }
321
322    /// Bind source and cache identity to the compiler-owned dependency graph.
323    ///
324    /// This mode requires the compiler's `dependencies` command and fails
325    /// closed on a missing, malformed, unsafe, or changing manifest. The
326    /// default remains entrypoint-only for compatibility with older compilers.
327    pub fn with_dependency_mode(mut self, mode: NativeTsDependencyMode) -> Self {
328        self.dependency_mode = mode;
329        self
330    }
331
332    /// Return the configured native TypeScript source-identity policy.
333    pub fn dependency_mode(&self) -> NativeTsDependencyMode {
334        self.dependency_mode
335    }
336
337    #[cfg(feature = "native-ts")]
338    /// Validate the workflow specification and ensure its native artifact is ready.
339    pub async fn preflight(&self, spec: &WorkflowSpec) -> Result<NativeTsRuntimePreflight> {
340        let (artifact, cache_hit) = self.compile_if_needed(spec).await?;
341        Ok(NativeTsRuntimePreflight {
342            entrypoint: artifact.entrypoint,
343            artifact: artifact.binary,
344            source_hash: artifact.source_hash,
345            cache_hit,
346        })
347    }
348
349    #[cfg(not(feature = "native-ts"))]
350    /// Return an error because native TypeScript support is not compiled in.
351    pub async fn preflight(&self, _spec: &WorkflowSpec) -> Result<NativeTsRuntimePreflight> {
352        Err(FlowError::Runtime(
353            "native-ts feature is disabled for NativeTsRuntime".to_string(),
354        ))
355    }
356
357    #[cfg(feature = "native-ts")]
358    async fn artifact_for(&self, spec: &WorkflowSpec) -> Result<NativeArtifact> {
359        validate_native_ts_spec(spec)?;
360        let (compiler_binary, compiler_fingerprint) = self
361            .compiler_identity_cache
362            .resolve_and_fingerprint(&self.config.compiler_binary)
363            .await?;
364        let working_dir = absolute_from_current_dir(&self.config.working_dir)?;
365        let entrypoint = resolve_against(&working_dir, &spec.runtime.entrypoint);
366        let cache_dir = absolute_from_current_dir(&self.config.cache_dir)?;
367        let (source_hash, source_snapshot, dependency_compiler_identity) = match self
368            .dependency_mode
369        {
370            NativeTsDependencyMode::EntrypointOnly => {
371                let (source_hash, source_snapshot) =
372                    SourceSnapshot::read(&entrypoint, spec).await?;
373                (source_hash, source_snapshot, None)
374            }
375            NativeTsDependencyMode::CompilerManifest => {
376                let graph = Box::pin(self.compiler_dependency_graph(
377                    &compiler_binary,
378                    &working_dir,
379                    &entrypoint,
380                ))
381                .await?;
382                let (source_hash, source_snapshot) =
383                    Box::pin(SourceSnapshot::read_dependency_graph(graph.sources, spec)).await?;
384                (source_hash, source_snapshot, Some(graph.compiler_identity))
385            }
386        };
387        // Keep the protocol-visible source hash portable, but scope the local
388        // native executable cache to every compile-environment input that can
389        // make identical workflow source produce an incompatible artifact.
390        let artifact_hash = native_artifact_cache_key(
391            &source_hash,
392            &compiler_binary,
393            &compiler_fingerprint,
394            &working_dir,
395            &entrypoint,
396            NATIVE_RUNTIME_PROTOCOL,
397            dependency_compiler_identity.as_deref(),
398        );
399        let name = format!("{}-{artifact_hash}", sanitize_filename(&spec.name));
400        let cache_entry = cache_dir.join(name);
401        Ok(NativeArtifact {
402            compiler_binary,
403            working_dir,
404            entrypoint,
405            binary: artifact_binary_path(&cache_entry),
406            cache_entry,
407            cache_key: artifact_hash,
408            source_hash,
409            source_snapshot,
410            dependency_compiler_identity,
411        })
412    }
413
414    #[cfg(feature = "native-ts")]
415    async fn compiler_dependency_graph(
416        &self,
417        compiler_binary: &Path,
418        working_dir: &Path,
419        entrypoint: &Path,
420    ) -> Result<VerifiedDependencyGraph> {
421        let child = Command::new(compiler_binary)
422            .arg("dependencies")
423            .arg(entrypoint)
424            .current_dir(working_dir)
425            .stdin(std::process::Stdio::null())
426            .stdout(std::process::Stdio::piped())
427            .stderr(std::process::Stdio::piped())
428            .kill_on_drop(true)
429            .spawn()?;
430        let output = communicate_with_bounded_output(
431            child,
432            "compiler dependency scan",
433            None,
434            MAX_DEPENDENCY_MANIFEST_BYTES,
435            self.max_stderr_bytes,
436            self.compile_timeout,
437        )
438        .await?;
439        if !output.status.success() {
440            return Err(FlowError::Runtime(format!(
441                "native TypeScript dependency scan failed: {}",
442                String::from_utf8_lossy(&output.stderr)
443            )));
444        }
445
446        verified_dependency_graph(&output.stdout, working_dir, entrypoint).await
447    }
448
449    #[cfg(feature = "native-ts")]
450    async fn compile_if_needed(&self, spec: &WorkflowSpec) -> Result<(NativeArtifact, bool)> {
451        let artifact = self.artifact_for(spec).await?;
452        match self
453            .artifact_cache
454            .inspect(&artifact.cache_entry, &artifact.cache_key)
455            .await?
456        {
457            ArtifactCacheState::Valid => return Ok((artifact, true)),
458            ArtifactCacheState::Missing => {}
459            ArtifactCacheState::Invalid(reason) => {
460                tracing::warn!(
461                    path = %artifact.cache_entry.display(),
462                    %reason,
463                    "repairing invalid native TypeScript cache entry"
464                );
465            }
466        }
467
468        let cache_dir = artifact.cache_entry.parent().ok_or_else(|| {
469            FlowError::Runtime(format!(
470                "native TypeScript cache entry {} has no parent directory",
471                artifact.cache_entry.display()
472            ))
473        })?;
474        tokio::fs::create_dir_all(cache_dir).await?;
475        // Bundle the executable and its integrity manifest in a unique cache
476        // entry. Renaming the directory is the atomic publish boundary, so a
477        // reader can observe either the complete entry or no entry at all.
478        let mut temporary_entry = TemporaryCacheEntryGuard::create(&artifact.cache_entry).await?;
479        let child = match Command::new(&artifact.compiler_binary)
480            .arg("compile")
481            .arg(&artifact.entrypoint)
482            .arg("-o")
483            .arg(temporary_entry.binary())
484            .current_dir(&artifact.working_dir)
485            .stdin(std::process::Stdio::null())
486            .stdout(std::process::Stdio::piped())
487            .stderr(std::process::Stdio::piped())
488            // A cancelled preflight must not leave the compiler running after
489            // its Rust future and temporary-artifact cleanup have disappeared.
490            .kill_on_drop(true)
491            .spawn()
492        {
493            Ok(child) => child,
494            Err(error) => {
495                temporary_entry.remove().await;
496                return Err(error.into());
497            }
498        };
499        let output = match communicate_with_bounded_output(
500            child,
501            "compiler",
502            None,
503            self.max_stdout_bytes,
504            self.max_stderr_bytes,
505            self.compile_timeout,
506        )
507        .await
508        {
509            Ok(output) => output,
510            Err(error) => {
511                temporary_entry.remove().await;
512                return Err(error);
513            }
514        };
515
516        if !output.status.success() {
517            temporary_entry.remove().await;
518            return Err(FlowError::Runtime(format!(
519                "native TypeScript compile failed: {}",
520                String::from_utf8_lossy(&output.stderr)
521            )));
522        }
523
524        let source_still_matches = match &artifact.dependency_compiler_identity {
525            None => artifact.source_snapshot.still_matches(spec).await,
526            Some(expected_identity) => {
527                match Box::pin(self.compiler_dependency_graph(
528                    &artifact.compiler_binary,
529                    &artifact.working_dir,
530                    &artifact.entrypoint,
531                ))
532                .await
533                {
534                    Ok(graph) if graph.compiler_identity == *expected_identity => {
535                        Box::pin(
536                            artifact
537                                .source_snapshot
538                                .still_matches_dependency_graph(graph.sources, spec),
539                        )
540                        .await
541                    }
542                    Ok(_) => Ok(false),
543                    Err(error) => Err(error),
544                }
545            }
546        };
547        match source_still_matches {
548            Ok(true) => {}
549            Ok(false) => {
550                temporary_entry.remove().await;
551                let subject = if artifact.dependency_compiler_identity.is_some() {
552                    "dependency graph or compiler identity".to_string()
553                } else {
554                    format!("source {}", artifact.entrypoint.display())
555                };
556                return Err(FlowError::Runtime(format!(
557                    "native TypeScript {subject} changed while it was being compiled; refusing to publish the artifact"
558                )));
559            }
560            Err(error) => {
561                temporary_entry.remove().await;
562                return Err(error);
563            }
564        }
565
566        if let Err(error) = self
567            .artifact_cache
568            .prepare(temporary_entry.path(), &artifact.cache_key)
569            .await
570        {
571            temporary_entry.remove().await;
572            return Err(error);
573        }
574        if let Err(error) = self
575            .artifact_cache
576            .publish(
577                temporary_entry.path(),
578                &artifact.cache_entry,
579                &artifact.cache_key,
580            )
581            .await
582        {
583            temporary_entry.remove().await;
584            return Err(error);
585        }
586        temporary_entry.disarm();
587
588        match self
589            .artifact_cache
590            .inspect(&artifact.cache_entry, &artifact.cache_key)
591            .await?
592        {
593            ArtifactCacheState::Valid => {}
594            ArtifactCacheState::Missing => {
595                return Err(FlowError::Runtime(format!(
596                    "native TypeScript cache entry {} disappeared after publication",
597                    artifact.cache_entry.display()
598                )));
599            }
600            ArtifactCacheState::Invalid(reason) => {
601                return Err(FlowError::Runtime(format!(
602                    "native TypeScript cache entry {} is invalid after publication: {reason}",
603                    artifact.cache_entry.display()
604                )));
605            }
606        }
607
608        Ok((artifact, false))
609    }
610
611    #[cfg(feature = "native-ts")]
612    async fn invoke<I, O>(
613        &self,
614        spec: &WorkflowSpec,
615        kind: NativeRuntimeKind,
616        payload: I,
617    ) -> Result<O>
618    where
619        I: Serialize + Send,
620        O: DeserializeOwned,
621    {
622        let (artifact, _) = self.compile_if_needed(spec).await?;
623        let request = serde_json::to_vec(&NativeRuntimeRequest::new(
624            kind,
625            spec.runtime.export_name.clone(),
626            artifact.source_hash,
627            payload,
628        ))?;
629
630        let child = Command::new(&artifact.binary)
631            .arg("--a3s-flow-runtime")
632            .stdin(std::process::Stdio::piped())
633            .stdout(std::process::Stdio::piped())
634            .stderr(std::process::Stdio::piped())
635            .current_dir(&artifact.working_dir)
636            // Boot timeouts, lease loss, shutdown, and caller cancellation all
637            // drop this future. Tie the direct artifact process to that owner.
638            .kill_on_drop(true)
639            .spawn()?;
640
641        let output = communicate_with_bounded_output(
642            child,
643            "runtime",
644            Some(request),
645            self.max_stdout_bytes,
646            self.max_stderr_bytes,
647            self.invocation_timeout,
648        )
649        .await?;
650        if !output.status.success() {
651            return Err(FlowError::Runtime(format!(
652                "native TypeScript runtime failed: {}",
653                String::from_utf8_lossy(&output.stderr)
654            )));
655        }
656
657        decode_native_response(kind, &output.stdout)
658    }
659}
660
661#[cfg(feature = "native-ts")]
662async fn communicate_with_bounded_output(
663    mut child: Child,
664    process_kind: &'static str,
665    stdin_bytes: Option<Vec<u8>>,
666    max_stdout_bytes: usize,
667    max_stderr_bytes: usize,
668    process_timeout: Option<Duration>,
669) -> Result<NativeProcessOutput> {
670    let stdin = match stdin_bytes {
671        Some(bytes) => Some((
672            child.stdin.take().ok_or_else(|| {
673                FlowError::Runtime(format!(
674                    "native TypeScript {process_kind} stdin pipe is unavailable"
675                ))
676            })?,
677            bytes,
678        )),
679        None => None,
680    };
681    let stdout = child.stdout.take().ok_or_else(|| {
682        FlowError::Runtime(format!(
683            "native TypeScript {process_kind} stdout pipe is unavailable"
684        ))
685    })?;
686    let stderr = child.stderr.take().ok_or_else(|| {
687        FlowError::Runtime(format!(
688            "native TypeScript {process_kind} stderr pipe is unavailable"
689        ))
690    })?;
691
692    let communication = collect_native_process_output(
693        &mut child,
694        stdin,
695        stdout,
696        stderr,
697        max_stdout_bytes,
698        max_stderr_bytes,
699    );
700    let output = match process_timeout {
701        Some(timeout) => match tokio::time::timeout(timeout, communication).await {
702            Ok(output) => output,
703            Err(_) => {
704                terminate_and_reap(&mut child).await;
705                return Err(FlowError::Runtime(format!(
706                    "native TypeScript {process_kind} timed out after {timeout:?}"
707                )));
708            }
709        },
710        None => communication.await,
711    };
712
713    match output {
714        Ok(output) => Ok(output),
715        Err(error) => {
716            // The read that crossed the limit stops consuming its pipe. Kill
717            // and reap the child so a blocked writer cannot outlive this call.
718            terminate_and_reap(&mut child).await;
719            match error {
720                NativeProcessOutputError::Io(error) => Err(error.into()),
721                NativeProcessOutputError::LimitExceeded { stream, limit } => {
722                    Err(FlowError::Runtime(format!(
723                        "native TypeScript {process_kind} {stream} exceeded the {limit}-byte limit"
724                    )))
725                }
726            }
727        }
728    }
729}
730
731#[cfg(feature = "native-ts")]
732async fn collect_native_process_output(
733    child: &mut Child,
734    stdin: Option<(ChildStdin, Vec<u8>)>,
735    stdout: ChildStdout,
736    stderr: ChildStderr,
737    max_stdout_bytes: usize,
738    max_stderr_bytes: usize,
739) -> std::result::Result<NativeProcessOutput, NativeProcessOutputError> {
740    let write_stdin = async move {
741        if let Some((mut stdin, bytes)) = stdin {
742            stdin
743                .write_all(&bytes)
744                .await
745                .map_err(NativeProcessOutputError::Io)?;
746            stdin
747                .shutdown()
748                .await
749                .map_err(NativeProcessOutputError::Io)?;
750        }
751        Ok(())
752    };
753    let wait = async { child.wait().await.map_err(NativeProcessOutputError::Io) };
754    let stdout = read_bounded_output(stdout, "stdout", max_stdout_bytes);
755    let stderr = read_bounded_output(stderr, "stderr", max_stderr_bytes);
756    let (status, (), stdout, stderr) = tokio::try_join!(wait, write_stdin, stdout, stderr)?;
757    Ok(NativeProcessOutput {
758        status,
759        stdout,
760        stderr,
761    })
762}
763
764#[cfg(feature = "native-ts")]
765async fn terminate_and_reap(child: &mut Child) {
766    let _ = child.start_kill();
767    let _ = child.wait().await;
768}
769
770#[cfg(feature = "native-ts")]
771async fn read_bounded_output<R>(
772    mut reader: R,
773    stream: &'static str,
774    limit: usize,
775) -> std::result::Result<Vec<u8>, NativeProcessOutputError>
776where
777    R: AsyncRead + Unpin,
778{
779    let mut output = Vec::with_capacity(limit.min(8 * 1024));
780    let mut buffer = [0_u8; 8 * 1024];
781    loop {
782        let count = reader
783            .read(&mut buffer)
784            .await
785            .map_err(NativeProcessOutputError::Io)?;
786        if count == 0 {
787            return Ok(output);
788        }
789        if count > limit.saturating_sub(output.len()) {
790            return Err(NativeProcessOutputError::LimitExceeded { stream, limit });
791        }
792        output.extend_from_slice(&buffer[..count]);
793    }
794}
795
796#[cfg(feature = "native-ts")]
797fn native_artifact_cache_key(
798    source_hash: &str,
799    compiler_binary: &Path,
800    compiler_fingerprint: &str,
801    working_dir: &Path,
802    entrypoint: &Path,
803    protocol: &str,
804    dependency_compiler_identity: Option<&str>,
805) -> String {
806    let legacy_identity = stable_hash([
807        b"a3s.flow.native_ts.artifact.v3".as_slice(),
808        source_hash.as_bytes(),
809        protocol.as_bytes(),
810        compiler_binary.as_os_str().as_encoded_bytes(),
811        compiler_fingerprint.as_bytes(),
812        working_dir.as_os_str().as_encoded_bytes(),
813        entrypoint.as_os_str().as_encoded_bytes(),
814        std::env::consts::OS.as_bytes(),
815        std::env::consts::ARCH.as_bytes(),
816    ]);
817    match dependency_compiler_identity {
818        None => legacy_identity,
819        Some(identity) => stable_hash([
820            b"a3s.flow.native_ts.artifact.compiler_identity.v1".as_slice(),
821            legacy_identity.as_bytes(),
822            identity.as_bytes(),
823        ]),
824    }
825}
826
827#[cfg(feature = "native-ts")]
828fn validate_native_ts_spec(spec: &WorkflowSpec) -> Result<()> {
829    spec.validate()?;
830    if spec.runtime.kind != RuntimeKind::NativeTs {
831        return Err(FlowError::InvalidWorkflow(format!(
832            "NativeTsRuntime requires a native_ts workflow spec, got {:?}",
833            spec.runtime.kind
834        )));
835    }
836    Ok(())
837}
838
839#[async_trait]
840impl FlowRuntime for NativeTsRuntime {
841    #[cfg(feature = "native-ts")]
842    async fn run_workflow(&self, invocation: WorkflowInvocation) -> Result<RuntimeCommand> {
843        let spec = invocation.spec.clone();
844        self.invoke(&spec, NativeRuntimeKind::Workflow, invocation)
845            .await
846    }
847
848    #[cfg(not(feature = "native-ts"))]
849    async fn run_workflow(&self, _invocation: WorkflowInvocation) -> Result<RuntimeCommand> {
850        Err(FlowError::Runtime(
851            "native-ts feature is disabled for NativeTsRuntime".to_string(),
852        ))
853    }
854
855    #[cfg(feature = "native-ts")]
856    async fn run_step(&self, invocation: StepInvocation) -> Result<JsonValue> {
857        let spec = workflow_spec_from_history(&invocation.history)?;
858        self.invoke(&spec, NativeRuntimeKind::Step, invocation)
859            .await
860    }
861
862    #[cfg(not(feature = "native-ts"))]
863    async fn run_step(&self, _invocation: StepInvocation) -> Result<JsonValue> {
864        Err(FlowError::Runtime(
865            "native-ts feature is disabled for NativeTsRuntime".to_string(),
866        ))
867    }
868}
869
870#[cfg(feature = "native-ts")]
871fn workflow_spec_from_history(history: &[FlowEventEnvelope]) -> Result<WorkflowSpec> {
872    let first = history
873        .first()
874        .ok_or_else(|| FlowError::Runtime("step invocation has empty history".to_string()))?;
875    match &first.event {
876        crate::model::FlowEvent::RunCreated { spec, .. } => Ok(spec.clone()),
877        _ => Err(FlowError::Runtime(
878            "first history event is not run_created".to_string(),
879        )),
880    }
881}
882
883#[cfg(feature = "native-ts")]
884fn sanitize_filename(value: &str) -> String {
885    value
886        .chars()
887        .map(|ch| {
888            if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
889                ch
890            } else {
891                '-'
892            }
893        })
894        .collect()
895}
896
897#[cfg(feature = "native-ts")]
898fn resolve_against(root: &Path, value: &str) -> PathBuf {
899    let path = PathBuf::from(value);
900    if path.is_absolute() {
901        path
902    } else {
903        root.join(path)
904    }
905}
906
907#[cfg(feature = "native-ts")]
908fn absolute_from_current_dir(path: &Path) -> Result<PathBuf> {
909    if path.is_absolute() {
910        return Ok(path.to_path_buf());
911    }
912    Ok(std::env::current_dir()?.join(path))
913}
914
915#[cfg(feature = "native-ts")]
916fn stable_hash(parts: impl IntoIterator<Item = impl AsRef<[u8]>>) -> String {
917    let mut hasher = Sha256::new();
918    for part in parts {
919        update_stable_hash_part(&mut hasher, part.as_ref());
920    }
921    hex_lower(&hasher.finalize())
922}
923
924#[cfg(feature = "native-ts")]
925fn update_stable_hash_part(hasher: &mut Sha256, bytes: &[u8]) {
926    hasher.update((bytes.len() as u64).to_le_bytes());
927    hasher.update(bytes);
928}
929
930#[cfg(feature = "native-ts")]
931fn hex_lower(bytes: &[u8]) -> String {
932    const HEX: &[u8; 16] = b"0123456789abcdef";
933    let mut output = String::with_capacity(bytes.len() * 2);
934    for byte in bytes {
935        output.push(HEX[(byte >> 4) as usize] as char);
936        output.push(HEX[(byte & 0x0f) as usize] as char);
937    }
938    output
939}
940
941#[cfg(feature = "native-ts")]
942fn decode_native_response<O>(kind: NativeRuntimeKind, bytes: &[u8]) -> Result<O>
943where
944    O: DeserializeOwned,
945{
946    let response: NativeRuntimeResponse = serde_json::from_slice(bytes)?;
947    if response.protocol != NATIVE_RUNTIME_PROTOCOL {
948        return Err(FlowError::Runtime(format!(
949            "native TypeScript runtime protocol mismatch: expected {NATIVE_RUNTIME_PROTOCOL}, got {}",
950            response.protocol
951        )));
952    }
953    if response.kind != kind {
954        return Err(FlowError::Runtime(format!(
955            "native TypeScript runtime response kind mismatch: expected {}, got {}",
956            kind.as_str(),
957            response.kind.as_str()
958        )));
959    }
960    if !response.ok {
961        let error = response
962            .error
963            .unwrap_or_else(|| "runtime returned ok=false without an error".to_string());
964        return Err(FlowError::Runtime(error));
965    }
966    let output = response.output.ok_or_else(|| {
967        FlowError::Runtime("native TypeScript runtime returned ok=true without output".to_string())
968    })?;
969    serde_json::from_value(output).map_err(FlowError::from)
970}
971
972#[cfg(test)]
973mod tests;
a3s_flow/runtime.rs

a3s_flow/
runtime.rs
