diff --git a/docker/kubernetes-tentacle/bootstrapRunner/bootstrapRunner.go b/docker/kubernetes-tentacle/bootstrapRunner/bootstrapRunner.go
index 0f7ce1007..f0550e0b3 100644
--- a/docker/kubernetes-tentacle/bootstrapRunner/bootstrapRunner.go
+++ b/docker/kubernetes-tentacle/bootstrapRunner/bootstrapRunner.go
@@ -6,9 +6,15 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"sync"
 	"time"
 )
 
+type SafeCounter struct {
+	Mutex sync.Mutex
+	Value int
+}
+
 // The bootstrapRunner applet is designed to execute a script in a specific folder
 // and format the script's output from stdout and stderr into the following format:
 // <line number>|<RFC3339Nano date time>|<"stdout" or "stderr">|<original string>
@@ -18,6 +24,9 @@ import (
 //
 // Note: all arguments given after the <script> argument are passed directly to the script as arguments.
 func main() {
+
+	lineCounter := SafeCounter{Value: 1, Mutex: sync.Mutex{}}
+
 	workspacePath := os.Args[1]
 	args := os.Args[2:]
 	cmd := exec.Command("bash", args[0:]...)
@@ -31,46 +40,23 @@ func main() {
 	doneStd := make(chan bool)
 	doneErr := make(chan bool)
 
-	//stdout log file
-	so, err := os.Create(workspacePath + "/stdout.log")
-	if err != nil {
-		panic(err)
-	}
-	// close fo on exit and check for its returned error
-	defer func() {
-		if err := so.Close(); err != nil {
-			panic(err)
-		}
-	}()
-	// make a write buffer
-	stdoutLogFile := bufio.NewWriter(so)
-
-	//stderr log file
-	se, err := os.Create(workspacePath + "/stderr.log")
-	if err != nil {
-		panic(err)
-	}
-	// close fo on exit and check for its returned error
-	defer func() {
-		if err := se.Close(); err != nil {
-			panic(err)
-		}
-	}()
-	// make a write buffer
-	stderrLogFile := bufio.NewWriter(se)
-
-	go reader(stdOutScanner, stdoutLogFile, &doneStd)
-	go reader(stdErrScanner, stderrLogFile, &doneErr)
-
-	err = cmd.Start()
-	if err != nil {
-		panic(err)
-	}
+	go reader(stdOutScanner, "stdout", &doneStd, &lineCounter)
+	go reader(stdErrScanner, "stderr", &doneErr, &lineCounter)
+
+	Write("stdout", "##octopus[stdout-verbose]", &lineCounter)
+	Write("stdout", "Kubernetes Script Pod started", &lineCounter)
+	Write("stdout", "##octopus[stdout-default]", &lineCounter)
+
+	err := cmd.Start()
 
 	// Wait for output buffering first
 	<-doneStd
 	<-doneErr
 
+	if err != nil {
+		panic(err)
+	}
+
 	err = cmd.Wait()
 
 	var exitErr *exec.ExitError
@@ -79,28 +65,32 @@ func main() {
 		fmt.Fprintln(os.Stderr, "bootstrapRunner.go: Failed to execute bootstrap script", err)
 	}
 
-	// Perform a final flush of the file buffers, just in case they didn't get flushed before
-	if err := stdoutLogFile.Flush(); err != nil {	
-		fmt.Fprintln(os.Stderr, "bootstrapRunner.go: Failed to perform final flush of stdoutLogFile", err)
-	}
-	if err := stderrLogFile.Flush(); err != nil {	
-		fmt.Fprintln(os.Stderr, "bootstrapRunner.go: Failed to perform final flush of stderrLogFile", err)
-	}
-	
-	os.Exit(cmd.ProcessState.ExitCode())
+	exitCode := cmd.ProcessState.ExitCode()
+
+	Write("stdout", "##octopus[stdout-verbose]", &lineCounter)
+	Write("stdout", "Kubernetes Script Pod completed", &lineCounter)
+	Write("stdout", "##octopus[stdout-default]", &lineCounter)
+
+	//TODO: Add this back to speed things up
+	//Write("stdout", fmt.Sprintf("EOS-075CD4F0-8C76-491D-BA76-0879D35E9CFE<<>>%d", exitCode))
+
+	os.Exit(exitCode)
 }
 
-func reader(scanner *bufio.Scanner, writer *bufio.Writer, done *chan bool) {
+func reader(scanner *bufio.Scanner, stream string, done *chan bool, counter *SafeCounter) {
 	for scanner.Scan() {
-		message := fmt.Sprintf("%s|%s\n", time.Now().UTC().Format(time.RFC3339Nano), scanner.Text())
-		fmt.Print(message)
-		if _, err := writer.WriteString(message); err != nil {
-			panic(err)
-		}
-
-		if err := writer.Flush(); err != nil {
-			panic(err)
-		}
+		Write(stream, scanner.Text(), counter)
 	}
 	*done <- true
 }
+
+func Write(stream string, text string, counter *SafeCounter) {
+	//Use a mutex to prevent race conditions updating the line number
+	//https://go.dev/tour/concurrency/9
+	counter.Mutex.Lock()
+
+	fmt.Printf("%d|%s|%s|%s\n", counter.Value, time.Now().UTC().Format(time.RFC3339Nano), stream, text)
+	counter.Value++
+
+	counter.Mutex.Unlock()
+}
diff --git a/source/Octopus.Tentacle.Tests/Kubernetes/KubernetesPodMonitorTests.cs b/source/Octopus.Tentacle.Tests/Kubernetes/KubernetesPodMonitorTests.cs
index ae89856be..670f63175 100644
--- a/source/Octopus.Tentacle.Tests/Kubernetes/KubernetesPodMonitorTests.cs
+++ b/source/Octopus.Tentacle.Tests/Kubernetes/KubernetesPodMonitorTests.cs
@@ -81,7 +81,7 @@ public async Task ExistingPodIsUpdatedWhenCompleted()
                 }
             };
 
-            podService.ListAllPodsAsync(Arg.Any<CancellationToken>())
+            podService.ListAllPods(Arg.Any<CancellationToken>())
                 .Returns(new V1PodList
                 {
                     Items = new List<V1Pod>
@@ -137,7 +137,7 @@ public async Task ExistingPodIsUpdatedWhenFailed()
                 }
             };
 
-            podService.ListAllPodsAsync(Arg.Any<CancellationToken>())
+            podService.ListAllPods(Arg.Any<CancellationToken>())
                 .Returns(new V1PodList
                 {
                     Items = new List<V1Pod>
@@ -199,7 +199,7 @@ public async Task StopTrackingPodWhenDeleted()
                 }
             };
 
-            podService.ListAllPodsAsync(Arg.Any<CancellationToken>())
+            podService.ListAllPods(Arg.Any<CancellationToken>())
                 .Returns(new V1PodList
                 {
                     Items = new List<V1Pod>
diff --git a/source/Octopus.Tentacle.Tests/Kubernetes/PodLogLineParserFixture.cs b/source/Octopus.Tentacle.Tests/Kubernetes/PodLogLineParserFixture.cs
new file mode 100644
index 000000000..c74623c9b
--- /dev/null
+++ b/source/Octopus.Tentacle.Tests/Kubernetes/PodLogLineParserFixture.cs
@@ -0,0 +1,89 @@
+using System;
+using System.Linq;
+using FluentAssertions;
+using NUnit.Framework;
+using Octopus.Tentacle.Contracts;
+using Octopus.Tentacle.Kubernetes;
+
+namespace Octopus.Tentacle.Tests.Kubernetes
+{
+    [TestFixture]
+    public class PodLogLineParserFixture
+    {
+        [TestCase("a|b|c", Reason = "Doesn't have 4 parts")]
+        public void NotCorrectlyPipeDelimited(string line)
+        {
+            var result = PodLogLineParser.ParseLine(line);
+            result.Error.Should().Contain("delimited");
+        }
+        
+        [TestCase("a|b|c|d", Reason = "Not a line number")]
+        public void FirstPartIsNotALineNumber(string line)
+        {
+            var result = PodLogLineParser.ParseLine(line);
+            result.Error.Should().Contain("line number");
+        }
+        
+        [TestCase("1|b|c|d", Reason = "Not a date")]
+        public void SecondPartIsNotALineDate(string line)
+        {
+            var result = PodLogLineParser.ParseLine(line);
+            result.Error.Should().Contain("DateTimeOffset");
+        }
+        
+        [TestCase("1|2024-04-03T06:03:10.501025551Z|c|d", Reason = "Not a valid source")]
+        public void ThirdPartIsNotAValidSource(string line)
+        {
+            var result = PodLogLineParser.ParseLine(line);
+            result.Error.Should().Contain("source");
+        }
+        
+        [Test]
+        public void SimpleMessage()
+        {
+            var logLine = PodLogLineParser.ParseLine("123|2024-04-03T06:03:10.501025551Z|stdout|This is the message").LogLine;
+            logLine.Should().NotBeNull();
+
+            logLine.LineNumber.Should().Be(123);
+            logLine.Source.Should().Be(ProcessOutputSource.StdOut);
+            logLine.Message.Should().Be("This is the message");
+            logLine.Occurred.Should().BeCloseTo(new DateTimeOffset(2024, 4, 3, 6, 3, 10, 501, TimeSpan.Zero), TimeSpan.FromMilliseconds(1));
+        }
+        
+        [Test]
+        public void ServiceMessage()
+        {
+            var logLine = PodLogLineParser.ParseLine("123|2024-04-03T06:03:10.501025551Z|stdout|##octopus[stdout-verbose]").LogLine;
+            logLine.Should().NotBeNull();
+
+            logLine.LineNumber.Should().Be(123);
+            logLine.Source.Should().Be(ProcessOutputSource.StdOut);
+            logLine.Message.Should().Be("##octopus[stdout-verbose]");
+            logLine.Occurred.Should().BeCloseTo(new DateTimeOffset(2024, 4, 3, 6, 3, 10, 501, TimeSpan.Zero), TimeSpan.FromMilliseconds(1));
+        }
+        
+        [Test]
+        public void ErrorMessage()
+        {
+            var logLine = PodLogLineParser.ParseLine("123|2024-04-03T06:03:10.501025551Z|stderr|Error!").LogLine;
+            logLine.Should().NotBeNull();
+
+            logLine.LineNumber.Should().Be(123);
+            logLine.Source.Should().Be(ProcessOutputSource.StdErr);
+            logLine.Message.Should().Be("Error!");
+            logLine.Occurred.Should().BeCloseTo(new DateTimeOffset(2024, 4, 3, 6, 3, 10, 501, TimeSpan.Zero), TimeSpan.FromMilliseconds(1));
+        }
+        
+        [Test]
+        public void MessageHasPipeInIt()
+        {
+            var logLine = PodLogLineParser.ParseLine("123|2024-04-03T06:03:10.501025551Z|stdout|This is the me|ss|age").LogLine;
+            logLine.Should().NotBeNull();
+
+            logLine.LineNumber.Should().Be(123);
+            logLine.Source.Should().Be(ProcessOutputSource.StdOut);
+            logLine.Message.Should().Be("This is the me|ss|age");
+            logLine.Occurred.Should().BeCloseTo(new DateTimeOffset(2024, 4, 3, 6, 3, 10, 501, TimeSpan.Zero), TimeSpan.FromMilliseconds(1));
+        }
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle.Tests/Kubernetes/PodLogReaderFixture.cs b/source/Octopus.Tentacle.Tests/Kubernetes/PodLogReaderFixture.cs
new file mode 100644
index 000000000..b086b6663
--- /dev/null
+++ b/source/Octopus.Tentacle.Tests/Kubernetes/PodLogReaderFixture.cs
@@ -0,0 +1,111 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using FluentAssertions;
+using NUnit.Framework;
+using Octopus.Tentacle.Contracts;
+using Octopus.Tentacle.Kubernetes;
+
+namespace Octopus.Tentacle.Tests.Kubernetes
+{
+    [TestFixture]
+    public class PodLogReaderFixture
+    {
+        [TestCase(0, Reason = "Initial position")]
+        [TestCase(4, Reason = "Subsequent position")]
+        [TestCase(12387126, Reason = "Large position")]
+        public async Task NoLines_SameSequenceNumber(long lastLogSequence)
+        {
+            string[] podLines = Array.Empty<string>();
+            
+            var reader = SetupReader(podLines);
+            var result = await PodLogReader.ReadPodLogs(lastLogSequence, reader);
+            result.NextSequenceNumber.Should().Be(lastLogSequence);
+            result.Lines.Should().BeEmpty();
+        }
+
+        [Test]
+        public async Task FirstLine_SequenceNumberIncreasesByOne()
+        {
+            string[] podLines = {
+                "1|2024-04-03T06:03:10.517865655Z|stdout|Kubernetes Script Pod completed",
+            };
+
+            var reader = SetupReader(podLines);
+            var result = await PodLogReader.ReadPodLogs(0, reader);
+            result.NextSequenceNumber.Should().Be(1);
+            result.Lines.Should().BeEquivalentTo(new[]
+            {
+                new ProcessOutput(ProcessOutputSource.StdOut, "Kubernetes Script Pod completed", DateTimeOffset.Parse("2024-04-03T06:03:10.517865655Z"))
+            });
+        }
+
+        [Test]
+        public async Task ThreeSubsequentLines_SequenceNumberIncreasesByThree()
+        {
+            string[] podLines = {
+                "5|2024-04-03T06:03:10.517857755Z|stdout|##octopus[stdout-verbose]",
+                "6|2024-04-03T06:03:10.517865655Z|stderr|Kubernetes Script Pod completed",
+                "7|2024-04-03T06:03:10.517867355Z|stdout|##octopus[stdout-default]"
+            };
+
+            var reader = SetupReader(podLines);
+            var result = await PodLogReader.ReadPodLogs(4, reader);
+            result.NextSequenceNumber.Should().Be(7);
+            result.Lines.Should().BeEquivalentTo(new[]
+            {
+                new ProcessOutput(ProcessOutputSource.StdOut, "##octopus[stdout-verbose]", DateTimeOffset.Parse("2024-04-03T06:03:10.517857755Z")),
+                new ProcessOutput(ProcessOutputSource.StdErr, "Kubernetes Script Pod completed", DateTimeOffset.Parse("2024-04-03T06:03:10.517865655Z")),
+                new ProcessOutput(ProcessOutputSource.StdOut, "##octopus[stdout-default]", DateTimeOffset.Parse("2024-04-03T06:03:10.517867355Z")),
+            });
+        }
+        
+        [Test]
+        public async Task StreamContainsPreviousLines_Deduplicates()
+        {
+            string[] podLines = {
+                "5|2024-04-03T06:03:10.517857755Z|stdout|##octopus[stdout-verbose]",
+                "6|2024-04-03T06:03:10.517865655Z|stderr|Kubernetes Script Pod completed",
+                "7|2024-04-03T06:03:10.517867355Z|stdout|##octopus[stdout-default]"
+            };
+
+            var allTaskLogs = new List<ProcessOutput>();
+            var reader = SetupReader(podLines.Take(1).ToArray());
+            var result = await PodLogReader.ReadPodLogs(4, reader);
+            result.NextSequenceNumber.Should().Be(5);
+            allTaskLogs.AddRange(result.Lines);
+            
+            reader = SetupReader(podLines.ToArray());
+            result = await PodLogReader.ReadPodLogs(5, reader);
+            result.NextSequenceNumber.Should().Be(7);
+            allTaskLogs.AddRange(result.Lines);
+            
+            allTaskLogs.Should().BeEquivalentTo(new[]
+            {
+                new ProcessOutput(ProcessOutputSource.StdOut, "##octopus[stdout-verbose]", DateTimeOffset.Parse("2024-04-03T06:03:10.517857755Z")),
+                new ProcessOutput(ProcessOutputSource.StdErr, "Kubernetes Script Pod completed", DateTimeOffset.Parse("2024-04-03T06:03:10.517865655Z")),
+                new ProcessOutput(ProcessOutputSource.StdOut, "##octopus[stdout-default]", DateTimeOffset.Parse("2024-04-03T06:03:10.517867355Z")),
+            });
+        }
+        
+        [Test]
+        public async Task MissingLine_Throws()
+        {
+            string[] podLines = {
+                "100|2024-04-03T06:03:10.517865655Z|stdout|Kubernetes Script Pod completed",
+            };
+        
+            var reader = SetupReader(podLines);
+            Func<Task> action = async () => await PodLogReader.ReadPodLogs(50, reader);
+            await action.Should().ThrowAsync<MissingPodLogException>();
+        }
+        
+        static StreamReader SetupReader(params string[] lines)
+        {
+            return new StreamReader(new MemoryStream(Encoding.Default.GetBytes(string.Join("\n", lines))));
+        }
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/IKubernetesPodLogService.cs b/source/Octopus.Tentacle/Kubernetes/IKubernetesPodLogService.cs
new file mode 100644
index 000000000..173b6ec03
--- /dev/null
+++ b/source/Octopus.Tentacle/Kubernetes/IKubernetesPodLogService.cs
@@ -0,0 +1,52 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
+using k8s.Autorest;
+using Octopus.Tentacle.Contracts;
+
+namespace Octopus.Tentacle.Kubernetes
+{
+    public interface IKubernetesPodLogService
+    {
+        Task<(IReadOnlyCollection<ProcessOutput>, long)> GetLogs(ScriptTicket scriptTicket, long lastLogSequence, CancellationToken cancellationToken);
+    }
+
+    class KubernetesPodLogService : KubernetesService, IKubernetesPodLogService
+    {
+        public KubernetesPodLogService(IKubernetesClientConfigProvider configProvider) : base(configProvider)
+        {
+        }
+
+        public async Task<(IReadOnlyCollection<ProcessOutput>, long)> GetLogs(ScriptTicket scriptTicket, long lastLogSequence, CancellationToken cancellationToken)
+        {
+            var podName = scriptTicket.ToKubernetesScriptPobName();
+            DateTimeOffset? sinceTime = null;
+
+            Stream logStream;
+
+            try
+            {
+                //TODO: Only grab recent
+                logStream = await Client.GetNamespacedPodLogsAsync(podName, KubernetesConfig.Namespace, podName, sinceTime, cancellationToken: cancellationToken);
+            }
+            catch (HttpOperationException ex)
+            {
+                //Pod logs aren't ready yet
+                if (ex.Response.StatusCode is HttpStatusCode.NotFound or HttpStatusCode.BadRequest)
+                {
+                    return (new List<ProcessOutput>(), lastLogSequence);
+                }
+
+                throw;
+            }
+            
+            using (var reader = new StreamReader(logStream))
+            {
+                return await PodLogReader.ReadPodLogs(lastLogSequence, reader);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/KubernetesClientExtensionMethods.cs b/source/Octopus.Tentacle/Kubernetes/KubernetesClientExtensionMethods.cs
new file mode 100644
index 000000000..00725c4fd
--- /dev/null
+++ b/source/Octopus.Tentacle/Kubernetes/KubernetesClientExtensionMethods.cs
@@ -0,0 +1,53 @@
+using System;
+using System.IO;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using k8s.Autorest;
+
+namespace Octopus.Tentacle.Kubernetes
+{
+    public static class KubernetesClientExtensionMethods
+    {
+        public static async Task<Stream> GetNamespacedPodLogsAsync(this k8s.Kubernetes client, string podName, string namespaceParameter, string container, DateTimeOffset? sinceTime = null, CancellationToken cancellationToken = default)
+        {
+            var url = $"api/v1/namespaces/{namespaceParameter}/pods/{podName}/log?container={container}";
+
+            if (sinceTime is not null)
+            {
+                var sinceTimeStr = sinceTime.Value.ToString("O");
+                url += $"&sinceTime={Uri.EscapeDataString(sinceTimeStr)}";
+            }
+
+            url = string.Concat(client.BaseUri, url);
+
+            var httpRequest = new HttpRequestMessage(HttpMethod.Get, url);
+
+            if (client.Credentials is not null)
+            {
+                await client.Credentials.ProcessHttpRequestAsync(httpRequest, CancellationToken.None);
+            }
+
+            var httpResponse = await client.HttpClient.SendAsync(httpRequest, HttpCompletionOption.ResponseHeadersRead, cancellationToken);
+
+            if (httpResponse.IsSuccessStatusCode)
+                return await httpResponse.Content.ReadAsStreamAsync();
+
+            // an exception occurred, throw
+            var responseContent = httpResponse.Content != null
+                ? await httpResponse.Content.ReadAsStringAsync().ConfigureAwait(false)
+                : string.Empty;
+
+            var ex = new HttpOperationException($"Operation returned an invalid status code '{httpResponse.StatusCode}', response body {responseContent}")
+            {
+                Request = new HttpRequestMessageWrapper(httpRequest, null),
+                Response = new HttpResponseMessageWrapper(httpResponse, responseContent)
+            };
+
+            httpRequest.Dispose();
+            httpResponse.Dispose();
+
+            throw ex;
+        }
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/KubernetesModule.cs b/source/Octopus.Tentacle/Kubernetes/KubernetesModule.cs
index 0e9540ecb..9808d5f38 100644
--- a/source/Octopus.Tentacle/Kubernetes/KubernetesModule.cs
+++ b/source/Octopus.Tentacle/Kubernetes/KubernetesModule.cs
@@ -1,6 +1,5 @@
 using Autofac;
 using Octopus.Tentacle.Background;
-using Octopus.Tentacle.Kubernetes.Scripts;
 
 namespace Octopus.Tentacle.Kubernetes
 {
@@ -14,9 +13,8 @@ protected override void Load(ContainerBuilder builder)
             builder.RegisterType<KubernetesConfigMapService>().As<IKubernetesConfigMapService>().SingleInstance();
             builder.RegisterType<KubernetesSecretService>().As<IKubernetesSecretService>().SingleInstance();
 
-            // this needs to be per-dependency, otherwise it re-uses the RunningKubernetesPod
-            builder.RegisterType<RunningKubernetesPod>().InstancePerDependency();
-            builder.RegisterType<KubernetesPodScriptExecutor>().AsSelf();
+            builder.RegisterType<KubernetesScriptPodCreator>().As<IKubernetesScriptPodCreator>().SingleInstance();
+            builder.RegisterType<KubernetesPodLogService>().As<IKubernetesPodLogService>().SingleInstance();
 
             builder.RegisterType<KubernetesPodMonitorTask>().As<IKubernetesPodMonitorTask>().As<IBackgroundTask>().SingleInstance();
             builder.RegisterType<KubernetesPodMonitor>().As<IKubernetesPodMonitor>().As<IKubernetesPodStatusProvider>().SingleInstance();
diff --git a/source/Octopus.Tentacle/Kubernetes/KubernetesPodMonitor.cs b/source/Octopus.Tentacle/Kubernetes/KubernetesPodMonitor.cs
index ac322e02e..6f91e3d99 100644
--- a/source/Octopus.Tentacle/Kubernetes/KubernetesPodMonitor.cs
+++ b/source/Octopus.Tentacle/Kubernetes/KubernetesPodMonitor.cs
@@ -74,7 +74,7 @@ internal async Task<string> InitialLoadAsync(CancellationToken cancellationToken
             log.Verbose("Preloading pod statuses");
 
             var newStatuses = new ConcurrentDictionary<ScriptTicket, TrackedScriptPod>();
-            var allPods = await podService.ListAllPodsAsync(cancellationToken);
+            var allPods = await podService.ListAllPods(cancellationToken);
             foreach (var pod in allPods.Items)
             {
                 var scriptTicket = pod.GetScriptTicket();
diff --git a/source/Octopus.Tentacle/Kubernetes/KubernetesPodService.cs b/source/Octopus.Tentacle/Kubernetes/KubernetesPodService.cs
index 6a3ec31f7..fcf8c4464 100644
--- a/source/Octopus.Tentacle/Kubernetes/KubernetesPodService.cs
+++ b/source/Octopus.Tentacle/Kubernetes/KubernetesPodService.cs
@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using k8s;
@@ -11,10 +12,11 @@ namespace Octopus.Tentacle.Kubernetes
     public interface IKubernetesPodService
     {
         Task<V1Pod?> TryGetPod(ScriptTicket scriptTicket, CancellationToken cancellationToken);
-        Task<V1PodList> ListAllPodsAsync(CancellationToken cancellationToken);
+        Task<V1PodList> ListAllPods(CancellationToken cancellationToken);
         Task WatchAllPods(string initialResourceVersion, Func<WatchEventType, V1Pod, CancellationToken, Task> onChange, Action<Exception> onError, CancellationToken cancellationToken);
         Task Create(V1Pod pod, CancellationToken cancellationToken);
         Task Delete(ScriptTicket scriptTicket, CancellationToken cancellationToken);
+        Task TryDelete(ScriptTicket commandScriptTicket, CancellationToken cancellationToken);
     }
 
     public class KubernetesPodService : KubernetesService, IKubernetesPodService
@@ -27,7 +29,7 @@ public KubernetesPodService(IKubernetesClientConfigProvider configProvider)
         public async Task<V1Pod?> TryGetPod(ScriptTicket scriptTicket, CancellationToken cancellationToken) =>
             await TryGetAsync(() => Client.ReadNamespacedPodAsync(scriptTicket.ToKubernetesScriptPobName(), KubernetesConfig.Namespace, cancellationToken: cancellationToken));
 
-        public async Task<V1PodList> ListAllPodsAsync(CancellationToken cancellationToken)
+        public async Task<V1PodList> ListAllPods(CancellationToken cancellationToken)
         {
             return await Client.ListNamespacedPodAsync(KubernetesConfig.Namespace,
                 labelSelector: OctopusLabels.ScriptTicketId,
@@ -64,7 +66,7 @@ public async Task WatchAllPods(string initialResourceVersion, Func<WatchEventTyp
             }
             catch (Exception ex)
             {
-                //Unfortunately we get an exception when the timeout hits (Server closes the connection) 
+                //Unfortunately we get an exception when the timeout hits (Server closes the connection)
                 //https://github.com/kubernetes-client/csharp/issues/828
                 if (ex is EndOfStreamException || ex.InnerException is EndOfStreamException)
                 {
@@ -76,7 +78,7 @@ public async Task WatchAllPods(string initialResourceVersion, Func<WatchEventTyp
                 }
             }
         }
-        
+
         public async Task Create(V1Pod pod, CancellationToken cancellationToken)
         {
             AddStandardMetadata(pod);
@@ -85,5 +87,8 @@ public async Task Create(V1Pod pod, CancellationToken cancellationToken)
 
         public async Task Delete(ScriptTicket scriptTicket, CancellationToken cancellationToken)
             => await Client.DeleteNamespacedPodAsync(scriptTicket.ToKubernetesScriptPobName(), KubernetesConfig.Namespace, cancellationToken: cancellationToken);
+
+        public async Task TryDelete(ScriptTicket scriptTicket, CancellationToken cancellationToken)
+            => await TryExecuteAsync(async () => await Delete(scriptTicket, cancellationToken));
     }
 }
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/KubernetesScriptPodCreator.cs b/source/Octopus.Tentacle/Kubernetes/KubernetesScriptPodCreator.cs
new file mode 100644
index 000000000..ff656e007
--- /dev/null
+++ b/source/Octopus.Tentacle/Kubernetes/KubernetesScriptPodCreator.cs
@@ -0,0 +1,234 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Security.Cryptography;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using k8s;
+using k8s.Models;
+using Newtonsoft.Json;
+using Octopus.Diagnostics;
+using Octopus.Tentacle.Configuration.Instances;
+using Octopus.Tentacle.Contracts.KubernetesScriptServiceV1Alpha;
+using Octopus.Tentacle.Kubernetes;
+using Octopus.Tentacle.Scripts;
+using Octopus.Tentacle.Variables;
+
+namespace Octopus.Tentacle.Kubernetes
+{
+    public interface IKubernetesScriptPodCreator
+    {
+        Task CreatePod(StartKubernetesScriptCommandV1Alpha command, IScriptWorkspace workspace, CancellationToken cancellationToken);
+    }
+
+    public class KubernetesScriptPodCreator : IKubernetesScriptPodCreator
+    {
+        readonly IKubernetesPodService podService;
+        readonly IKubernetesSecretService secretService;
+        readonly IKubernetesPodContainerResolver containerResolver;
+        readonly IApplicationInstanceSelector appInstanceSelector;
+        readonly ISystemLog log;
+
+        public KubernetesScriptPodCreator(
+            IKubernetesPodService podService,
+            IKubernetesSecretService secretService,
+            IKubernetesPodContainerResolver containerResolver,
+            IApplicationInstanceSelector appInstanceSelector,
+            ISystemLog log)
+        {
+            this.podService = podService;
+            this.secretService = secretService;
+            this.containerResolver = containerResolver;
+            this.appInstanceSelector = appInstanceSelector;
+            this.log = log;
+        }
+
+        public async Task CreatePod(StartKubernetesScriptCommandV1Alpha command, IScriptWorkspace workspace, CancellationToken cancellationToken)
+        {
+            using (ScriptIsolationMutex.Acquire(workspace.IsolationLevel,
+                       workspace.ScriptMutexAcquireTimeout,
+                       workspace.ScriptMutexName ?? nameof(KubernetesScriptPodCreator),
+                       message =>
+                       {
+                       },
+                       command.TaskId,
+                       cancellationToken,
+                       log))
+            {
+                //Possibly create the image pull secret name
+                var imagePullSecretName = await CreateImagePullSecret(command, cancellationToken);
+
+                //create the k8s pod
+                await CreatePod(command, workspace, imagePullSecretName, cancellationToken);
+            }
+        }
+
+        async Task<string?> CreateImagePullSecret(StartKubernetesScriptCommandV1Alpha command, CancellationToken cancellationToken)
+        {
+            //if we have no feed url or no username, then we can't create image secrets
+            if (command.PodImageConfiguration?.FeedUrl is null || command.PodImageConfiguration?.FeedUsername is null)
+                return null;
+
+            var secretName = CreateImagePullSecretName(command.PodImageConfiguration.FeedUrl, command.PodImageConfiguration.FeedUsername);
+
+            // this structure is a docker config auth file
+            // https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/#inspecting-the-secret-regcred
+            var config = new Dictionary<string, object>
+            {
+                ["auths"] = new Dictionary<string, object>
+                {
+                    [command.PodImageConfiguration.FeedUrl] = new
+                    {
+                        username = command.PodImageConfiguration.FeedUsername,
+                        password = command.PodImageConfiguration.FeedPassword,
+                        auth = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{command.PodImageConfiguration.FeedUsername}:{command.PodImageConfiguration.FeedPassword}"))
+                    }
+                }
+            };
+
+            var configJson = JsonConvert.SerializeObject(config, Formatting.None);
+            var secretData = new Dictionary<string, byte[]>
+            {
+                [".dockerconfigjson"] = Encoding.UTF8.GetBytes(configJson)
+            };
+
+            var existingSecret = await secretService.TryGetSecretAsync(secretName, cancellationToken);
+            if (existingSecret is null)
+            {
+                //if there is no secret, create one
+                var secret = new V1Secret
+                {
+                    Type = "kubernetes.io/dockerconfigjson",
+                    Metadata = new V1ObjectMeta
+                    {
+                        Name = secretName,
+                        NamespaceProperty = KubernetesConfig.Namespace
+                    },
+                    Data = new Dictionary<string, byte[]>
+                    {
+                        [".dockerconfigjson"] = Encoding.UTF8.GetBytes(configJson)
+                    }
+                };
+
+                await secretService.CreateSecretAsync(secret, cancellationToken);
+            }
+            else
+            {
+                //patch the existing secret with the data (just in case the password has changed for this feed/user combo)
+                await secretService.UpdateSecretDataAsync(secretName, secretData, cancellationToken);
+            }
+
+            return secretName;
+        }
+
+        static string CreateImagePullSecretName(string feedUrl, string? username)
+        {
+            //The secret name is the domain of the feed & the username in a hash
+            //We use SHA1 because we want a small hash (as a secrets name is max 253 chars) and we aren't hashing secure data
+            using var sha1 = SHA1.Create();
+            var feedUri = new Uri(feedUrl);
+
+            var hash = Convert.ToBase64String(sha1.ComputeHash(Encoding.UTF8.GetBytes($"{feedUri.DnsSafeHost}:{username}")));
+
+            //remove all special chars from the hash
+            var sanitizedHash = new string(hash.Where(c => c != '/' && c != '+' && c != '=').ToArray());
+
+            //the secret name must be lowercase. We lose a bit of uniqueness as base64 is case-sensitive
+            //but the downside of a clash isn't _that_ bad (just overwriting an existing secret)
+            return $"octopus-feed-cred-{sanitizedHash}".ToLowerInvariant();
+        }
+
+        async Task CreatePod(StartKubernetesScriptCommandV1Alpha command, IScriptWorkspace workspace, string? imagePullSecretName, CancellationToken cancellationToken)
+        {
+            var podName = command.ScriptTicket.ToKubernetesScriptPobName();
+
+            log.Verbose($"Creating Kubernetes Pod '{podName}'.");
+
+            //write the bootstrap runner script to the workspace
+            workspace.CopyFile(KubernetesConfig.BootstrapRunnerExecutablePath, "bootstrapRunner");
+
+            var scriptName = Path.GetFileName(workspace.BootstrapScriptFilePath);
+
+            //Deserialize the volume configuration from the environment configuration
+            var volumes = KubernetesJson.Deserialize<List<V1Volume>>(KubernetesConfig.PodVolumeJson);
+            var pod = new V1Pod
+            {
+                Metadata = new V1ObjectMeta
+                {
+                    Name = podName,
+                    NamespaceProperty = KubernetesConfig.Namespace,
+                    Labels = new Dictionary<string, string>
+                    {
+                        ["octopus.com/serverTaskId"] = command.TaskId,
+                        ["octopus.com/scriptTicketId"] = command.ScriptTicket.TaskId
+                    }
+                },
+                Spec = new V1PodSpec
+                {
+                    Containers = new List<V1Container>
+                    {
+                        new()
+                        {
+                            Name = podName,
+                            Image = command.PodImageConfiguration?.Image ?? await containerResolver.GetContainerImageForCluster(),
+                            Command = new List<string> { $"/octopus/Work/{command.ScriptTicket.TaskId}/bootstrapRunner" },
+                            Args = new List<string>
+                                {
+                                    $"/octopus/Work/{command.ScriptTicket.TaskId}",
+                                    $"/octopus/Work/{command.ScriptTicket.TaskId}/{scriptName}"
+                                }.Concat(workspace.ScriptArguments ?? Array.Empty<string>())
+                                .ToList(),
+                            VolumeMounts = new List<V1VolumeMount>
+                            {
+                                new("/octopus", "tentacle-home"),
+                            },
+                            Env = new List<V1EnvVar>
+                            {
+                                new(KubernetesConfig.NamespaceVariableName, KubernetesConfig.Namespace),
+                                new(KubernetesConfig.HelmReleaseNameVariableName, KubernetesConfig.HelmReleaseName),
+                                new(KubernetesConfig.HelmChartVersionVariableName, KubernetesConfig.HelmChartVersion),
+                                new(EnvironmentVariables.TentacleHome, $"/octopus"),
+                                new(EnvironmentVariables.TentacleInstanceName, appInstanceSelector.Current.InstanceName),
+                                new(EnvironmentVariables.TentacleVersion, Environment.GetEnvironmentVariable(EnvironmentVariables.TentacleVersion)),
+                                new(EnvironmentVariables.TentacleCertificateSignatureAlgorithm, Environment.GetEnvironmentVariable(EnvironmentVariables.TentacleCertificateSignatureAlgorithm)),
+                                new("OCTOPUS_RUNNING_IN_CONTAINER", "Y")
+
+                                //We intentionally exclude setting "TentacleJournal" since it doesn't make sense to keep a Deployment Journal for Kubernetes deployments
+                            },
+                            Resources = new V1ResourceRequirements
+                            {
+                                //set resource requests to be quite low for now as the scripts tend to run fairly quickly
+                                Requests = new Dictionary<string, ResourceQuantity>
+                                {
+                                    ["cpu"] = new("25m"),
+                                    ["memory"] = new("100Mi")
+                                }
+                            }
+                        }
+                    },
+                    //only include the image pull secret name if it's actually been defined
+                    ImagePullSecrets = imagePullSecretName is not null
+                        ? new List<V1LocalObjectReference>
+                        {
+                            new(imagePullSecretName)
+                        }
+                        : new List<V1LocalObjectReference>(),
+                    ServiceAccountName = KubernetesConfig.PodServiceAccountName,
+                    RestartPolicy = "Never",
+                    Volumes = volumes,
+                    //currently we only support running on linux nodes
+                    NodeSelector = new Dictionary<string, string>
+                    {
+                        ["kubernetes.io/os"] = "linux"
+                    }
+                }
+            };
+
+            await podService.Create(pod, cancellationToken);
+
+            log.Verbose($"Executing script in Kubernetes Pod '{podName}'.");
+        }
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/KubernetesService.cs b/source/Octopus.Tentacle/Kubernetes/KubernetesService.cs
index 905bdb181..8a9a9ad83 100644
--- a/source/Octopus.Tentacle/Kubernetes/KubernetesService.cs
+++ b/source/Octopus.Tentacle/Kubernetes/KubernetesService.cs
@@ -48,5 +48,16 @@ protected void AddStandardMetadata(IKubernetesObject<V1ObjectMeta> k8sObject)
                 return null;
             }
         }
+
+        protected static async Task TryExecuteAsync(Func<Task> action)
+        {
+            try
+            {
+                await action();
+            }
+            catch (HttpOperationException opException)
+                when (opException.Response.StatusCode == HttpStatusCode.NotFound)
+            { }
+        }
     }
 }
diff --git a/source/Octopus.Tentacle/Kubernetes/PodLogLineParser.cs b/source/Octopus.Tentacle/Kubernetes/PodLogLineParser.cs
new file mode 100644
index 000000000..ba4c78feb
--- /dev/null
+++ b/source/Octopus.Tentacle/Kubernetes/PodLogLineParser.cs
@@ -0,0 +1,68 @@
+using System;
+using Octopus.Tentacle.Contracts;
+
+namespace Octopus.Tentacle.Kubernetes
+{
+    public record PodLogLine(int LineNumber, ProcessOutputSource Source, string Message, DateTimeOffset Occurred);
+
+    public class PodLogParseResult
+    {
+        public bool Succeeded { get; }
+        public PodLogLine? LogLine { get; }
+        public string? Error { get; }
+
+        PodLogParseResult(bool succeeded, PodLogLine? logLine, string? error)
+        {
+            Succeeded = succeeded;
+            LogLine = logLine;
+            Error = error;
+        }
+
+        public static PodLogParseResult Success(PodLogLine logLine)
+        {
+            return new PodLogParseResult(true, logLine, null);
+        }
+        
+        public static PodLogParseResult Fail(string error)
+        {
+            return new PodLogParseResult(false, null, error);
+        }
+    }
+
+    static class PodLogLineParser
+    {
+        public static PodLogParseResult ParseLine(string line)
+        {
+            var logParts = line.Split(new[] { '|' }, 4);
+
+            if (logParts.Length != 4)
+            {
+                return PodLogParseResult.Fail($"Invalid log line detected. '{line}' is not correctly pipe-delimited.");
+            }
+
+            if (!int.TryParse(logParts[0], out int lineNumber))
+            {
+                return PodLogParseResult.Fail($"Invalid log line detected. '{logParts[0]}' is not a valid line number.");
+            }
+
+            if (!DateTimeOffset.TryParse(logParts[1], out var occurred))
+            {
+                return PodLogParseResult.Fail($"Invalid log line detected. Failed to parse '{logParts[1]}' as a DateTimeOffset.");
+            }
+
+            if (!Enum.TryParse(logParts[2], true, out ProcessOutputSource source))
+            {
+                return PodLogParseResult.Fail($"Invalid log line detected. '{logParts[2]}' is not a valid source.");
+            }
+            
+            //add the new line
+            var message = logParts[3];
+
+            //TODO: confirm if this line is useful
+            // if (message.StartsWith("EOS-075CD4F0-8C76-491D-BA76-0879D35E9CFE"))
+            //     source = ProcessOutputSource.Debug;
+            
+            return PodLogParseResult.Success(new PodLogLine(lineNumber, source, message, occurred));
+        }
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/PodLogReader.cs b/source/Octopus.Tentacle/Kubernetes/PodLogReader.cs
new file mode 100644
index 000000000..123f624ff
--- /dev/null
+++ b/source/Octopus.Tentacle/Kubernetes/PodLogReader.cs
@@ -0,0 +1,54 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Threading.Tasks;
+using Octopus.Tentacle.Contracts;
+using Octopus.Tentacle.Util;
+
+namespace Octopus.Tentacle.Kubernetes
+{
+    static class PodLogReader
+    {
+        public static async Task<(IReadOnlyCollection<ProcessOutput> Lines, long NextSequenceNumber)> ReadPodLogs(long lastLogSequence, StreamReader reader)
+        {
+            var results = new List<ProcessOutput>();
+            var nextSequenceNumber = lastLogSequence;
+            bool haveSeenPodLogEntryMatchingLogSequence = false;
+            
+            while (true)
+            {
+                var line = await reader.ReadLineAsync();
+
+                //No more to read
+                if (line.IsNullOrEmpty())
+                {
+                    return (results, nextSequenceNumber);
+                }
+
+                //TODO: print parse errors
+                var parseResult = PodLogLineParser.ParseLine(line!);
+                var podLogLine = parseResult.LogLine;
+                
+                //Pod log line numbers are 1-based, log sequence is 0-based
+                if (podLogLine != null && podLogLine.LineNumber > lastLogSequence)
+                {
+                    if (podLogLine.LineNumber == lastLogSequence + 1)
+                        haveSeenPodLogEntryMatchingLogSequence = true;
+
+                    if (!haveSeenPodLogEntryMatchingLogSequence)
+                        throw new MissingPodLogException();
+
+                    results.Add(new ProcessOutput(podLogLine.Source, podLogLine.Message, podLogLine.Occurred));
+                    nextSequenceNumber = podLogLine.LineNumber;
+                }
+
+                //TODO: try not to read any more if we see a panic?
+                    
+            }
+        }
+    }
+
+    class MissingPodLogException : Exception
+    {
+    }
+}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/Scripts/KubernetesPodOutputStreamWriter.cs b/source/Octopus.Tentacle/Kubernetes/Scripts/KubernetesPodOutputStreamWriter.cs
deleted file mode 100644
index ba837ab4e..000000000
--- a/source/Octopus.Tentacle/Kubernetes/Scripts/KubernetesPodOutputStreamWriter.cs
+++ /dev/null
@@ -1,144 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using Octopus.Tentacle.Contracts;
-using Octopus.Tentacle.Scripts;
-using Octopus.Tentacle.Util;
-
-namespace Octopus.Tentacle.Kubernetes.Scripts
-{
-    public class KubernetesPodOutputStreamWriter
-    {
-        readonly IScriptWorkspace workspace;
-        long lastStdOutOffset;
-        long lastStdErrOffset;
-
-        public KubernetesPodOutputStreamWriter(IScriptWorkspace workspace)
-        {
-            this.workspace = workspace;
-        }
-
-        public async Task StreamPodLogsToScriptLog(IScriptLogWriter writer, CancellationToken cancellationToken, bool isFinalRead = false)
-        {
-            try
-            {
-                //open the file streams for reading
-                using var stdOutStream = await SafelyOpenLogStreamReader("stdout.log", cancellationToken);
-                using var stdErrStream = await SafelyOpenLogStreamReader("stderr.log", cancellationToken);
-
-                //if either of these is null, just return
-                if (stdOutStream is null || stdErrStream is null)
-                    return;
-
-                // This loop is exited when either the cancellation token is cancelled (which is when the pod is finished or the script is cancelled)
-                // or if this is the final read, at the end (so we read once and jump out)
-                while (true)
-                {
-                    if (cancellationToken.IsCancellationRequested)
-                        break;
-
-                    //Read both the stdout and stderr log files
-                    var stdOutReadTask = ReadLogFileTail(writer, stdOutStream, ProcessOutputSource.StdOut, lastStdOutOffset);
-                    var stdErrReadTask = ReadLogFileTail(writer, stdErrStream, ProcessOutputSource.StdErr, lastStdErrOffset);
-
-                    //wait for them to both complete
-                    await Task.WhenAll(stdOutReadTask, stdErrReadTask);
-
-                    //store the final offsets
-                    lastStdOutOffset = stdOutReadTask.Result.FinalOffset;
-                    lastStdErrOffset = stdErrReadTask.Result.FinalOffset;
-
-                    //stitch the log lines together and order by occurred, then write to actual log
-                    var orderedLogLines = stdOutReadTask.Result.Logs
-                        .Concat(stdErrReadTask.Result.Logs)
-                        .OrderBy(ll => ll.Occurred);
-
-                    //write all the read log lines to the output script log
-                    foreach (var logLine in orderedLogLines)
-                    {
-                        writer.WriteOutput(logLine.Source, logLine.Message, logLine.Occurred);
-                    }
-
-                    //wait for 250ms before reading the logs again (except on the final read)
-                    if (!isFinalRead)
-                    {
-                        await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken);
-                    }
-                    else
-                    {
-                        //if this is the last read we need to jump out (and not spin forever)
-                        break;
-                    }
-                }
-            }
-            catch (TaskCanceledException)
-            {
-                //ignore all task cancelled exceptions as they may be thrown by the pod finishing (and thus signally)
-            }
-        }
-
-        async Task<StreamReader?> SafelyOpenLogStreamReader(string filename, CancellationToken cancellationToken)
-        {
-            while (true)
-            {
-                if (cancellationToken.IsCancellationRequested)
-                    return null;
-
-                try
-                {
-                    //we expect that we will receive a number of FileNotFoundException's while the pod is spinning up
-                    //Eventually the file should exist
-                    return new StreamReader(workspace.OpenFileStreamForReading(filename), Encoding.UTF8);
-                }
-                catch (FileNotFoundException)
-                {
-                    //wait for 50ms before reading the logs again
-                    await Task.Delay(TimeSpan.FromMilliseconds(50), cancellationToken);
-                }
-            }
-        }
-
-        record LogLine(ProcessOutputSource Source, string Message, DateTimeOffset Occurred);
-
-        static async Task<(IEnumerable<LogLine> Logs, long FinalOffset)> ReadLogFileTail(IScriptLogWriter writer, StreamReader reader, ProcessOutputSource source, long lastOffset)
-        {
-            if (reader.BaseStream.Length == lastOffset)
-                return (Enumerable.Empty<LogLine>(), lastOffset);
-
-            reader.BaseStream.Seek(lastOffset, SeekOrigin.Begin);
-
-            var newLines = new List<LogLine>();
-            string? line;
-            do
-            {
-                line = await reader.ReadLineAsync();
-                if (line.IsNullOrEmpty())
-                    break;
-
-                var logParts = line!.Split(new[] { '|' }, 2);
-
-                if (logParts.Length != 2)
-                {
-                    writer.WriteOutput(ProcessOutputSource.StdErr, $"Invalid log line detected. '{line}' is not correctly pipe-delimited.");
-                    continue;
-                }
-
-                //part 1 is the datetimeoffset
-                if (!DateTimeOffset.TryParse(logParts[0], out var occurred))
-                {
-                    writer.WriteOutput(ProcessOutputSource.StdErr, $"Failed to parse '{logParts[0]}' as a DateTimeOffset. Using DateTimeOffset.UtcNow.");
-                    occurred = DateTimeOffset.UtcNow;
-                }
-
-                //add the new line
-                newLines.Add(new LogLine(source, logParts[1], occurred));
-            } while (!line.IsNullOrEmpty());
-
-            return (newLines, reader.BaseStream.Position);
-        }
-    }
-}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/Scripts/KubernetesPodScriptExecutor.cs b/source/Octopus.Tentacle/Kubernetes/Scripts/KubernetesPodScriptExecutor.cs
deleted file mode 100644
index 82d20d809..000000000
--- a/source/Octopus.Tentacle/Kubernetes/Scripts/KubernetesPodScriptExecutor.cs
+++ /dev/null
@@ -1,32 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using Octopus.Tentacle.Contracts.KubernetesScriptServiceV1Alpha;
-using Octopus.Tentacle.Scripts;
-
-namespace Octopus.Tentacle.Kubernetes.Scripts
-{
-    public class KubernetesPodScriptExecutor
-    {
-        readonly RunningKubernetesPod.Factory runningKubernetesPodFactory;
-
-        public KubernetesPodScriptExecutor(RunningKubernetesPod.Factory runningKubernetesPodFactory)
-        {
-            this.runningKubernetesPodFactory = runningKubernetesPodFactory;
-        }
-
-        public IRunningScript ExecuteOnBackgroundThread(StartKubernetesScriptCommandV1Alpha command, IScriptWorkspace workspace, ScriptStateStore scriptStateStore, CancellationToken cancellationToken)
-        {
-            var runningScript = runningKubernetesPodFactory(
-                workspace,
-                workspace.CreateLog(),
-                command,
-                scriptStateStore,
-                cancellationToken);
-
-            Task.Run(() => runningScript.Execute(), cancellationToken);
-
-            return runningScript;
-        }
-    }
-}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Kubernetes/Scripts/RunningKubernetesPod.cs b/source/Octopus.Tentacle/Kubernetes/Scripts/RunningKubernetesPod.cs
deleted file mode 100644
index 69a0d75d2..000000000
--- a/source/Octopus.Tentacle/Kubernetes/Scripts/RunningKubernetesPod.cs
+++ /dev/null
@@ -1,467 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Security.Cryptography;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using k8s;
-using k8s.Models;
-using Newtonsoft.Json;
-using Octopus.Diagnostics;
-using Octopus.Tentacle.Configuration.Instances;
-using Octopus.Tentacle.Contracts;
-using Octopus.Tentacle.Contracts.KubernetesScriptServiceV1Alpha;
-using Octopus.Tentacle.Scripts;
-using Octopus.Tentacle.Util;
-using Octopus.Tentacle.Variables;
-
-namespace Octopus.Tentacle.Kubernetes.Scripts
-{
-    public class RunningKubernetesPod : IRunningScript
-    {
-        public delegate RunningKubernetesPod Factory(
-            IScriptWorkspace workspace,
-            IScriptLog scriptLog,
-            StartKubernetesScriptCommandV1Alpha command,
-            IScriptStateStore stateStore,
-            CancellationToken scriptCancellationToken);
-
-        readonly IScriptWorkspace workspace;
-        readonly StartKubernetesScriptCommandV1Alpha command;
-        readonly ScriptTicket scriptTicket;
-        readonly string taskId;
-        readonly ILog log;
-        readonly IScriptStateStore stateStore;
-        readonly IKubernetesPodService podService;
-        readonly IKubernetesPodStatusProvider podStatusProvider;
-        readonly IKubernetesSecretService secretService;
-        readonly IKubernetesPodContainerResolver containerResolver;
-        readonly CancellationToken scriptCancellationToken;
-        readonly string? instanceName;
-        readonly KubernetesPodOutputStreamWriter outputStreamWriter;
-        readonly string podName;
-
-        public int ExitCode { get; private set; }
-        public ProcessState State { get; private set; }
-        public IScriptLog ScriptLog { get; }
-
-        public RunningKubernetesPod(IScriptWorkspace workspace,
-            IScriptLog scriptLog,
-            StartKubernetesScriptCommandV1Alpha command,
-            IScriptStateStore stateStore,
-            CancellationToken scriptCancellationToken,
-            ISystemLog log,
-            IKubernetesPodService podService,
-            IKubernetesPodStatusProvider podStatusProvider,
-            IKubernetesSecretService secretService,
-            IKubernetesPodContainerResolver containerResolver,
-            IApplicationInstanceSelector appInstanceSelector)
-        {
-            this.workspace = workspace;
-            this.command = command;
-            scriptTicket = command.ScriptTicket;
-            taskId = command.TaskId;
-            this.log = log;
-            this.stateStore = stateStore;
-            this.podService = podService;
-            this.podStatusProvider = podStatusProvider;
-            this.secretService = secretService;
-            this.containerResolver = containerResolver;
-            this.scriptCancellationToken = scriptCancellationToken;
-            ScriptLog = scriptLog;
-            instanceName = appInstanceSelector.Current.InstanceName;
-
-            outputStreamWriter = new KubernetesPodOutputStreamWriter(workspace);
-
-            State = ProcessState.Pending;
-
-            // this doesn't change, so build it once
-            podName = scriptTicket.ToKubernetesScriptPobName();
-        }
-
-        public async Task Execute()
-        {
-            var exitCode = -1;
-
-            try
-            {
-                using var writer = ScriptLog.CreateWriter();
-
-                //register a cancellation callback so that when the script is cancelled, we cancel the pod
-                //we use a using to make sure this callback is deregistered
-                using var cancellationTokenRegistration = scriptCancellationToken.Register(() =>
-                {
-                    //we spawn the pod cancellation on a background thread (as this callback runs synchronously)
-                    Task.Run(async () =>
-                    {
-                        try
-                        {
-                            WriteVerbose(writer, $"Deleting Kubernetes Pod '{podName}'.");
-                            //We delete the pob (because we no longer need it)
-                            await podService.Delete(scriptTicket, CancellationToken.None);
-                            WriteVerbose(writer, $"Deleted Kubernetes Pod '{podName}'.");
-                        }
-                        catch (Exception e)
-                        {
-                            WriteError(writer, $"Failed to delete Kubernetes Pod {podName}. {e}");
-                        }
-                    }, CancellationToken.None);
-                });
-
-                try
-                {
-                    using (ScriptIsolationMutex.Acquire(workspace.IsolationLevel,
-                               workspace.ScriptMutexAcquireTimeout,
-                               workspace.ScriptMutexName ?? nameof(RunningKubernetesPod),
-                               message => writer.WriteOutput(ProcessOutputSource.StdOut, message),
-                               taskId,
-                               scriptCancellationToken,
-                               log))
-                    {
-                        //Possibly create the image pull secret name
-                        var imagePullSecretName = await CreateImagePullSecret(scriptCancellationToken);
-
-                        //create the k8s pod
-                        await CreatePod(writer, imagePullSecretName, scriptCancellationToken);
-
-                        State = ProcessState.Running;
-                        RecordScriptHasStarted(writer);
-
-                        //we now need to monitor the resulting pod status
-                        exitCode = await MonitorPodAndLogs(writer);
-                    }
-                }
-                catch (OperationCanceledException)
-                {
-                    WriteInfo(writer, "Script execution canceled.");
-                    exitCode = ScriptExitCodes.CanceledExitCode;
-                }
-                catch (TimeoutException)
-                {
-                    WriteInfo(writer, "Script execution timed out.");
-                    exitCode = ScriptExitCodes.TimeoutExitCode;
-                }
-            }
-            catch (Exception ex)
-            {
-                log.Error(ex, $"Failed to execute script {scriptTicket.TaskId} in pod {podName}");
-                exitCode = ScriptExitCodes.FatalExitCode;
-            }
-            finally
-            {
-                try
-                {
-                    RecordScriptHasCompleted(exitCode);
-                }
-                finally
-                {
-                    ExitCode = exitCode;
-                    State = ProcessState.Complete;
-                }
-            }
-        }
-
-        async Task<int> MonitorPodAndLogs(IScriptLogWriter writer)
-        {
-            var podCompletionCancellationTokenSource = new CancellationTokenSource();
-            var checkPodTask = CheckIfPodHasCompleted(podCompletionCancellationTokenSource);
-
-            //we pass the pod completion CTS here because its used to cancel the writing of the pod stream
-            var monitorPodOutputTask = outputStreamWriter.StreamPodLogsToScriptLog(writer, podCompletionCancellationTokenSource.Token);
-
-            await Task.WhenAll(checkPodTask, monitorPodOutputTask);
-
-            //once they have both finished, perform one last log read (and don't cancel on it)
-            await outputStreamWriter.StreamPodLogsToScriptLog(writer, CancellationToken.None, true);
-
-            //return the exit code of the pod
-            return checkPodTask.Result;
-        }
-
-        async Task<int> CheckIfPodHasCompleted(CancellationTokenSource podCompletionCancellationTokenSource)
-        {
-            var resultStatusCode = ScriptExitCodes.UnknownScriptExitCode;
-            ITrackedScriptPod? status = null;
-            while (!scriptCancellationToken.IsCancellationRequested)
-            {
-                status = podStatusProvider.TryGetTrackedScriptPod(scriptTicket);
-                if (status is not null && status.State == TrackedScriptPodState.Succeeded)
-                {
-                    resultStatusCode = 0;
-                    break;
-                }
-
-                if (status is not null && status.State == TrackedScriptPodState.Failed)
-                {
-                    resultStatusCode = status.ExitCode!.Value;
-                    break;
-                }
-
-                await Task.Delay(250, scriptCancellationToken);
-            }
-
-            //if the pod was killed by cancellation, then we need to change the exit code
-            if (scriptCancellationToken.IsCancellationRequested)
-            {
-                resultStatusCode = ScriptExitCodes.CanceledExitCode;
-            }
-
-            podCompletionCancellationTokenSource.Cancel();
-
-            log.Verbose($"Pod {podName} completed.{status}");
-
-            return resultStatusCode;
-        }
-
-        async Task<string?> CreateImagePullSecret(CancellationToken cancellationToken)
-        {
-            //if we have no feed url or no username, then we can't create image secrets
-            if (command.PodImageConfiguration?.FeedUrl is null || command.PodImageConfiguration?.FeedUsername is null)
-                return null;
-
-            var secretName = CreateImagePullSecretName(command.PodImageConfiguration.FeedUrl, command.PodImageConfiguration.FeedUsername);
-
-            // this structure is a docker config auth file
-            // https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/#inspecting-the-secret-regcred
-            var config = new Dictionary<string, object>
-            {
-                ["auths"] = new Dictionary<string, object>
-                {
-                    [command.PodImageConfiguration.FeedUrl] = new
-                    {
-                        username = command.PodImageConfiguration.FeedUsername,
-                        password = command.PodImageConfiguration.FeedPassword,
-                        auth = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{command.PodImageConfiguration.FeedUsername}:{command.PodImageConfiguration.FeedPassword}"))
-                    }
-                }
-            };
-
-            var configJson = JsonConvert.SerializeObject(config, Formatting.None);
-            var secretData = new Dictionary<string, byte[]>
-            {
-                [".dockerconfigjson"] = Encoding.UTF8.GetBytes(configJson)
-            };
-
-            var existingSecret = await secretService.TryGetSecretAsync(secretName, cancellationToken);
-            if (existingSecret is null)
-            {
-                //if there is no secret, create one
-                var secret = new V1Secret
-                {
-                    Type = "kubernetes.io/dockerconfigjson",
-                    Metadata = new V1ObjectMeta
-                    {
-                        Name = secretName,
-                        NamespaceProperty = KubernetesConfig.Namespace
-                    },
-                    Data = new Dictionary<string, byte[]>
-                    {
-                        [".dockerconfigjson"] = Encoding.UTF8.GetBytes(configJson)
-                    }
-                };
-
-                await secretService.CreateSecretAsync(secret, cancellationToken);
-            }
-            else
-            {
-                //patch the existing secret with the data (just in case the password has changed for this feed/user combo)
-                await secretService.UpdateSecretDataAsync(secretName, secretData, cancellationToken);
-            }
-
-            return secretName;
-        }
-
-        static string CreateImagePullSecretName(string feedUrl, string? username)
-        {
-            //The secret name is the domain of the feed & the username in a hash
-            //We use SHA1 because we want a small hash (as a secrets name is max 253 chars) and we aren't hashing secure data
-            using var sha1 = SHA1.Create();
-            var feedUri = new Uri(feedUrl);
-
-            var hash = Convert.ToBase64String(sha1.ComputeHash(Encoding.UTF8.GetBytes($"{feedUri.DnsSafeHost}:{username}")));
-
-            //remove all special chars from the hash
-            var sanitizedHash = new string(hash.Where(c => c != '/' && c != '+' && c != '=').ToArray());
-
-            //the secret name must be lowercase. We lose a bit of uniqueness as base64 is case-sensitive
-            //but the downside of a clash isn't _that_ bad (just overwriting an existing secret)
-            return $"octopus-feed-cred-{sanitizedHash}".ToLowerInvariant();
-        }
-
-        async Task CreatePod(IScriptLogWriter writer, string? imagePullSecretName, CancellationToken cancellationToken)
-        {
-            WriteVerbose(writer, $"Creating Kubernetes Pod '{podName}'.");
-
-            //write the bootstrap runner script to the workspace
-            workspace.CopyFile(KubernetesConfig.BootstrapRunnerExecutablePath, "bootstrapRunner");
-
-            var scriptName = Path.GetFileName(workspace.BootstrapScriptFilePath);
-
-            //Deserialize the volume configuration from the environment configuration
-            var volumes = KubernetesJson.Deserialize<List<V1Volume>>(KubernetesConfig.PodVolumeJson);
-
-            var pod = new V1Pod
-            {
-                Metadata = new V1ObjectMeta
-                {
-                    Name = podName,
-                    NamespaceProperty = KubernetesConfig.Namespace,
-                    Labels = new Dictionary<string, string>
-                    {
-                        ["octopus.com/serverTaskId"] = taskId,
-                        ["octopus.com/scriptTicketId"] = scriptTicket.TaskId
-                    }
-                },
-                Spec = new V1PodSpec
-                {
-                    Containers = new List<V1Container>
-                    {
-                        new()
-                        {
-                            Name = podName,
-                            Image = command.PodImageConfiguration?.Image ?? await containerResolver.GetContainerImageForCluster(),
-                            Command = new List<string> { $"/octopus/Work/{scriptTicket.TaskId}/bootstrapRunner" },
-                            Args = new List<string>
-                                {
-                                    $"/octopus/Work/{scriptTicket.TaskId}",
-                                    $"/octopus/Work/{scriptTicket.TaskId}/{scriptName}"
-                                }.Concat(workspace.ScriptArguments ?? Array.Empty<string>())
-                                .ToList(),
-                            VolumeMounts = new List<V1VolumeMount>
-                            {
-                                new("/octopus", "tentacle-home"),
-                            },
-                            Env = new List<V1EnvVar>
-                            {
-                                new(KubernetesConfig.NamespaceVariableName, KubernetesConfig.Namespace),
-                                new(KubernetesConfig.HelmReleaseNameVariableName, KubernetesConfig.HelmReleaseName),
-                                new(KubernetesConfig.HelmChartVersionVariableName, KubernetesConfig.HelmChartVersion),
-                                new(EnvironmentVariables.TentacleHome, $"/octopus"),
-                                new(EnvironmentVariables.TentacleInstanceName, instanceName),
-                                new(EnvironmentVariables.TentacleVersion, Environment.GetEnvironmentVariable(EnvironmentVariables.TentacleVersion)),
-                                new(EnvironmentVariables.TentacleCertificateSignatureAlgorithm, Environment.GetEnvironmentVariable(EnvironmentVariables.TentacleCertificateSignatureAlgorithm)),
-                                new("OCTOPUS_RUNNING_IN_CONTAINER", "Y")
-
-                                //We intentionally exclude setting "TentacleJournal" since it doesn't make sense to keep a Deployment Journal for Kubernetes deployments
-                            },
-                            Resources = new V1ResourceRequirements
-                            {
-                                //set resource requests to be quite low for now as the scripts tend to run fairly quickly
-                                Requests = new Dictionary<string, ResourceQuantity>
-                                {
-                                    ["cpu"] = new("25m"),
-                                    ["memory"] = new("100Mi")
-                                }
-                            }
-                        }
-                    },
-                    //only include the image pull secret name if it's actually been defined
-                    ImagePullSecrets = imagePullSecretName is not null
-                        ? new List<V1LocalObjectReference>
-                        {
-                            new(imagePullSecretName)
-                        }
-                        : new List<V1LocalObjectReference>(),
-                    ServiceAccountName = KubernetesConfig.PodServiceAccountName,
-                    RestartPolicy = "Never",
-                    Volumes = volumes,
-                    //currently we only support running on linux nodes
-                    NodeSelector = new Dictionary<string, string>
-                    {
-                        ["kubernetes.io/os"] = "linux"
-                    }
-                }
-            };
-
-            await podService.Create(pod, cancellationToken);
-
-            WriteVerbose(writer, $"Executing script in Kubernetes Pod '{podName}'.");
-        }
-
-        public async Task Cleanup(CancellationToken cancellationToken)
-        {
-            if (KubernetesConfig.DisableAutomaticPodCleanup)
-            {
-                log.Verbose($"Not deleting completed pod {podName} as automatic cleanup is disabled.");
-                return;
-            }
-
-            try
-            {
-                log.Verbose($"Deleting completed pod {podName}.");
-                await podService.Delete(scriptTicket, cancellationToken);
-            }
-            catch (Exception ex)
-            {
-                //we can't write this back to the script log as it's already cleaned up at this point
-                log.Error(ex, $"Failed to delete Pod {podName}.");
-            }
-        }
-
-        void RecordScriptHasStarted(IScriptLogWriter writer)
-        {
-            try
-            {
-                var scriptState = stateStore.Load();
-                scriptState.Start();
-                stateStore.Save(scriptState);
-            }
-            catch (Exception ex)
-            {
-                try
-                {
-                    WriteInfo(writer, $"Warning: An exception occurred saving the ScriptState for pod '{podName}': {ex.Message}");
-                    WriteInfo(writer, ex.ToString());
-                }
-                catch
-                {
-                    //we don't care about errors here
-                }
-            }
-        }
-
-        void RecordScriptHasCompleted(int exitCode)
-        {
-            using var writer = ScriptLog.CreateWriter();
-            try
-            {
-                var scriptState = stateStore.Load();
-                scriptState.Complete(exitCode);
-                stateStore.Save(scriptState);
-                WriteVerbose(writer, $"Kubernetes Pod '{podName}' completed with exit code {exitCode}");
-            }
-            catch (Exception ex)
-            {
-                try
-                {
-                    WriteInfo(writer, $"Warning: An exception occurred saving the ScriptState for pod '{podName}': {ex.Message}");
-                    WriteInfo(writer, ex.ToString());
-                }
-                catch
-                {
-                    //we don't care about errors here
-                }
-            }
-        }
-
-        void WriteInfo(IScriptLogWriter writer, string message)
-        {
-            writer.WriteOutput(ProcessOutputSource.StdOut, message);
-            log.Info(message);
-        }
-
-        void WriteError(IScriptLogWriter writer, string message)
-        {
-            writer.WriteOutput(ProcessOutputSource.StdErr, message);
-            log.Error(message);
-        }
-
-        void WriteVerbose(IScriptLogWriter writer, string message)
-        {
-            writer.WriteVerbose(message);
-            log.Verbose(message);
-        }
-    }
-}
\ No newline at end of file
diff --git a/source/Octopus.Tentacle/Services/Scripts/Kubernetes/KubernetesScriptServiceV1Alpha.cs b/source/Octopus.Tentacle/Services/Scripts/Kubernetes/KubernetesScriptServiceV1Alpha.cs
index 039996023..993d1b47f 100644
--- a/source/Octopus.Tentacle/Services/Scripts/Kubernetes/KubernetesScriptServiceV1Alpha.cs
+++ b/source/Octopus.Tentacle/Services/Scripts/Kubernetes/KubernetesScriptServiceV1Alpha.cs
@@ -1,10 +1,13 @@
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Octopus.Diagnostics;
 using Octopus.Tentacle.Contracts;
 using Octopus.Tentacle.Contracts.KubernetesScriptServiceV1Alpha;
-using Octopus.Tentacle.Kubernetes.Scripts;
+using Octopus.Tentacle.Kubernetes;
 using Octopus.Tentacle.Maintenance;
 using Octopus.Tentacle.Scripts;
 using Octopus.Tentacle.Util;
@@ -14,171 +17,126 @@ namespace Octopus.Tentacle.Services.Scripts.Kubernetes
     [KubernetesService(typeof(IKubernetesScriptServiceV1Alpha))]
     public class KubernetesScriptServiceV1Alpha : IAsyncKubernetesScriptServiceV1Alpha, IRunningScriptReporter
     {
-        readonly Lazy<KubernetesPodScriptExecutor> executor;
+        readonly IKubernetesPodService podService;
         readonly IScriptWorkspaceFactory workspaceFactory;
-        readonly IScriptStateStoreFactory scriptStateStoreFactory;
-        readonly ConcurrentDictionary<ScriptTicket, RunningScriptWrapper> runningScripts = new();
+        readonly IKubernetesPodStatusProvider podStatusProvider;
+        readonly IKubernetesScriptPodCreator podCreator;
+        readonly IKubernetesPodLogService logService;
+        readonly ISystemLog log;
+
+        //TODO: check what will happen when Tentacle restarts
+        readonly ConcurrentDictionary<ScriptTicket, Lazy<SemaphoreSlim>> startScriptMutexes = new();
 
         public KubernetesScriptServiceV1Alpha(
-            Lazy<KubernetesPodScriptExecutor> executor,
+            IKubernetesPodService podService,
             IScriptWorkspaceFactory workspaceFactory,
-            IScriptStateStoreFactory scriptStateStoreFactory)
+            IKubernetesPodStatusProvider podStatusProvider,
+            IKubernetesScriptPodCreator podCreator,
+            IKubernetesPodLogService logService,
+            ISystemLog log)
         {
-            this.executor = executor;
+            this.podService = podService;
             this.workspaceFactory = workspaceFactory;
-            this.scriptStateStoreFactory = scriptStateStoreFactory;
+            this.podStatusProvider = podStatusProvider;
+            this.podCreator = podCreator;
+            this.logService = logService;
+            this.log = log;
         }
 
         public async Task<KubernetesScriptStatusResponseV1Alpha> StartScriptAsync(StartKubernetesScriptCommandV1Alpha command, CancellationToken cancellationToken)
         {
-            var runningScript = runningScripts.GetOrAdd(
-                command.ScriptTicket,
-                _ =>
-                {
-                    var workspace = workspaceFactory.GetWorkspace(command.ScriptTicket);
-                    var scriptState = scriptStateStoreFactory.Create(workspace);
-                    return new RunningScriptWrapper(scriptState);
-                });
-
-            using (await runningScript.StartScriptMutex.LockAsync(cancellationToken))
+            var mutex = startScriptMutexes.GetOrAdd(command.ScriptTicket, _ => new Lazy<SemaphoreSlim>(() => new SemaphoreSlim(1, 1))).Value;
+            using (await mutex.LockAsync(cancellationToken))
             {
-                IScriptWorkspace workspace;
-
-                // If the state already exists then this runningScript is already running/has already run and we should not run it again
-                if (runningScript.ScriptStateStore.Exists())
+                var trackedPod = podStatusProvider.TryGetTrackedScriptPod(command.ScriptTicket);
+                if (trackedPod != null)
                 {
-                    var state = runningScript.ScriptStateStore.Load();
-
-                    if (state.HasStarted() || runningScript.Process != null)
-                    {
-                        return await GetResponse(command.ScriptTicket, 0, runningScript.Process);
-                    }
-
-                    workspace = workspaceFactory.GetWorkspace(command.ScriptTicket);
-                }
-                else
-                {
-                    workspace = await workspaceFactory.PrepareWorkspace(command.ScriptTicket,
-                        command.ScriptBody,
-                        command.Scripts,
-                        command.Isolation,
-                        command.ScriptIsolationMutexTimeout,
-                        command.IsolationMutexName,
-                        command.Arguments,
-                        command.Files,
-                        cancellationToken);
-
-                    runningScript.ScriptStateStore.Create();
+                    return await GetResponse(trackedPod, 0, cancellationToken);
                 }
 
-                var process = executor.Value.ExecuteOnBackgroundThread(command, workspace, runningScript.ScriptStateStore, runningScript.CancellationToken);
+                //TODO: consider adding an idempotent version of PrepareWorkspace
+                var workspace = await workspaceFactory.PrepareWorkspace(command.ScriptTicket,
+                    command.ScriptBody,
+                    command.Scripts,
+                    command.Isolation,
+                    command.ScriptIsolationMutexTimeout,
+                    command.IsolationMutexName,
+                    command.Arguments,
+                    command.Files,
+                    cancellationToken);
+
+                //create the pod
+                await podCreator.CreatePod(command, workspace, cancellationToken);
 
-                runningScript.Process = process;
+                var (logs, _) = await logService.GetLogs(command.ScriptTicket, 0, cancellationToken);
 
-                return await GetResponse(command.ScriptTicket, 0, runningScript.Process);
+                //return a status that say's we are pending
+                return new KubernetesScriptStatusResponseV1Alpha(command.ScriptTicket, ProcessState.Pending, 0, logs.ToList(), 0);
             }
         }
 
         public async Task<KubernetesScriptStatusResponseV1Alpha> GetStatusAsync(KubernetesScriptStatusRequestV1Alpha request, CancellationToken cancellationToken)
         {
-            runningScripts.TryGetValue(request.ScriptTicket, out var runningScript);
-            return await GetResponse(request.ScriptTicket, request.LastLogSequence, runningScript?.Process);
+            await Task.CompletedTask;
+
+            var trackedPod = podStatusProvider.TryGetTrackedScriptPod(request.ScriptTicket);
+            return trackedPod != null
+                ? await GetResponse(trackedPod, request.LastLogSequence, cancellationToken)
+                //if we are getting the status of an unknown pod, return that it's still pending
+                : new KubernetesScriptStatusResponseV1Alpha(request.ScriptTicket, ProcessState.Pending, 0, new List<ProcessOutput>(), request.LastLogSequence);
         }
 
         public async Task<KubernetesScriptStatusResponseV1Alpha> CancelScriptAsync(CancelKubernetesScriptCommandV1Alpha command, CancellationToken cancellationToken)
         {
-            if (runningScripts.TryGetValue(command.ScriptTicket, out var runningScript))
-            {
-                runningScript.Cancel();
-            }
+            var trackedPod = podStatusProvider.TryGetTrackedScriptPod(command.ScriptTicket);
+            //if we are cancelling a pod that doesn't exist, just return complete with an unknown script exit code
+            if (trackedPod == null)
+                return new KubernetesScriptStatusResponseV1Alpha(command.ScriptTicket, ProcessState.Complete, ScriptExitCodes.UnknownScriptExitCode, new List<ProcessOutput>(), command.LastLogSequence);
 
-            return await GetResponse(command.ScriptTicket, command.LastLogSequence, runningScript?.Process);
+            var response = await GetResponse(trackedPod, command.LastLogSequence, cancellationToken);
+
+            //delete the pod
+            await podService.Delete(command.ScriptTicket, cancellationToken);
+
+            return response;
         }
 
         public async Task CompleteScriptAsync(CompleteKubernetesScriptCommandV1Alpha command, CancellationToken cancellationToken)
         {
-            if (runningScripts.TryRemove(command.ScriptTicket, out var runningScript))
-            {
-                runningScript.Dispose();
-            }
+            startScriptMutexes.TryRemove(command.ScriptTicket, out _);
 
             var workspace = workspaceFactory.GetWorkspace(command.ScriptTicket);
             await workspace.Delete(cancellationToken);
 
-            if (runningScript?.Process is not null)
-                await runningScript.Process.Cleanup(cancellationToken);
+            //we do a try delete as the cancel might have already deleted it
+            if (!KubernetesConfig.DisableAutomaticPodCleanup)
+                await podService.TryDelete(command.ScriptTicket, cancellationToken);
         }
 
-        async Task<KubernetesScriptStatusResponseV1Alpha> GetResponse(ScriptTicket ticket, long lastLogSequence, IRunningScript? runningScript)
+        async Task<KubernetesScriptStatusResponseV1Alpha> GetResponse(ITrackedScriptPod trackedPod, long lastLogSequence, CancellationToken cancellationToken)
         {
-            await Task.CompletedTask;
-
-            var workspace = workspaceFactory.GetWorkspace(ticket);
-            var scriptLog = runningScript?.ScriptLog ?? workspace.CreateLog();
-            var logs = scriptLog.GetOutput(lastLogSequence, out var next);
-
-            if (runningScript != null)
-            {
-                return new KubernetesScriptStatusResponseV1Alpha(ticket, runningScript.State, runningScript.ExitCode, logs, next);
-            }
-
-            // If we don't have a RunningProcess we check the ScriptStateStore to see if we have persisted a script result
-            var scriptStateStore = scriptStateStoreFactory.Create(workspace);
-            if (scriptStateStore.Exists())
+            var processState = trackedPod.State switch
             {
-                var scriptState = scriptStateStore.Load();
-
-                if (!scriptState.HasCompleted())
-                {
-                    scriptState.Complete(ScriptExitCodes.UnknownResultExitCode, false);
-                    scriptStateStore.Save(scriptState);
-                }
-
-                return new KubernetesScriptStatusResponseV1Alpha(ticket, scriptState.State, scriptState.ExitCode ?? ScriptExitCodes.UnknownResultExitCode, logs, next);
-            }
-
-            return new KubernetesScriptStatusResponseV1Alpha(ticket, ProcessState.Complete, ScriptExitCodes.UnknownScriptExitCode, logs, next);
+                TrackedScriptPodState.Running => ProcessState.Running,
+                TrackedScriptPodState.Succeeded => ProcessState.Complete,
+                TrackedScriptPodState.Failed => ProcessState.Complete,
+                _ => throw new ArgumentOutOfRangeException()
+            };
+
+            var (outputLogs, nextLogSequence) = await logService.GetLogs(trackedPod.ScriptTicket, lastLogSequence, cancellationToken);
+
+            return new KubernetesScriptStatusResponseV1Alpha(
+                trackedPod.ScriptTicket,
+                processState,
+                trackedPod.ExitCode ?? 0,
+                outputLogs.ToList(),
+                nextLogSequence
+            );
         }
 
         public bool IsRunningScript(ScriptTicket ticket)
         {
-            if (runningScripts.TryGetValue(ticket, out var script))
-            {
-                if (script.Process?.State != ProcessState.Complete)
-                {
-                    return true;
-                }
-            }
-
-            return false;
-        }
-
-        class RunningScriptWrapper : IDisposable
-        {
-            readonly CancellationTokenSource cancellationTokenSource = new();
-
-            public RunningScriptWrapper(ScriptStateStore scriptStateStore)
-            {
-                ScriptStateStore = scriptStateStore;
-
-                CancellationToken = cancellationTokenSource.Token;
-            }
-
-            public IRunningScript? Process { get; set; }
-            public ScriptStateStore ScriptStateStore { get; }
-            public SemaphoreSlim StartScriptMutex { get; } = new(1, 1);
-
-            public CancellationToken CancellationToken { get; }
-
-            public void Cancel()
-            {
-                cancellationTokenSource.Cancel();
-            }
-
-            public void Dispose()
-            {
-                cancellationTokenSource.Dispose();
-            }
+            return podStatusProvider.TryGetTrackedScriptPod(ticket) is not null;
         }
     }
 }
\ No newline at end of file