Implement a background service that auto-syncs package ops

  between instances

Author: OceanOak

backend/migrations/20251119_000000_multi_branch_sync.sql

@@ -0,0 +1,7 @@
+
+ALTER TABLE package_ops ADD COLUMN instance_id TEXT NULL
+  REFERENCES instances(id) ON DELETE SET NULL;
+
+-- Create index for querying ops by instance
+CREATE INDEX IF NOT EXISTS idx_package_ops_instance
+  ON package_ops(instance_id) WHERE instance_id IS NOT NULL;

backend/src/BuiltinCli/Builtin.fs

@@ -14,6 +14,7 @@ let builtins =
       Libs.File.builtins
       Libs.Execution.builtins
       Libs.Output.builtins
+      Libs.Process.builtins
       Libs.Stdin.builtins
       Libs.Time.builtins
       Libs.Terminal.builtins ]

backend/src/BuiltinCli/BuiltinCli.fsproj

@@ -14,6 +14,7 @@
     <Compile Include="Libs/File.fs" />
     <Compile Include="Libs/Output.fs" />
     <Compile Include="Libs/Execution.fs" />
+    <Compile Include="Libs/Process.fs" />
     <Compile Include="Libs/Stdin.fs" />
     <Compile Include="Libs/Time.fs" />
     <Compile Include="Libs/Terminal.fs" />

backend/src/BuiltinCli/Libs/Process.fs

@@ -0,0 +1,145 @@
+/// Standard libraries for process management
+module BuiltinCli.Libs.Process
+
+open System.Threading.Tasks
+open FSharp.Control.Tasks
+
+open Prelude
+open LibExecution.RuntimeTypes
+
+module Dval = LibExecution.Dval
+module Builtin = LibExecution.Builtin
+open Builtin.Shortcuts
+
+let fns : List<BuiltInFn> =
+  [ { name = fn "processSpawnBackground" 0
+      typeParams = []
+      parameters =
+        [ Param.make "args" (TList TString) "Arguments to pass to the CLI" ]
+      returnType = TypeReference.result TInt64 TString
+      description =
+        "Spawns the current CLI executable in the background with the given arguments. Returns the process ID (PID) on success."
+      fn =
+        (function
+        | _state, _, _, [ DList(_vtTODO, args) ] ->
+          uply {
+            try
+              let argStrings =
+                args
+                |> List.map (fun arg ->
+                  match arg with
+                  | DString s -> s
+                  | _ -> Exception.raiseInternal "Expected string arguments" [])
+
+              // Get the current executable path
+              let currentExe =
+                System.Diagnostics.Process.GetCurrentProcess().MainModule.FileName
+
+              let psi = System.Diagnostics.ProcessStartInfo()
+              psi.FileName <- currentExe
+              psi.UseShellExecute <- false
+              psi.CreateNoWindow <- true
+
+              // Add arguments
+              for arg in argStrings do
+                psi.ArgumentList.Add(arg)
+
+              let proc = System.Diagnostics.Process.Start(psi)
+
+              if isNull proc then
+                return
+                  Dval.resultError
+                    KTInt64
+                    KTString
+                    (DString "Failed to start background process")
+              else
+                return Dval.resultOk KTInt64 KTString (DInt64(int64 proc.Id))
+            with ex ->
+              return
+                Dval.resultError
+                  KTInt64
+                  KTString
+                  (DString $"Error spawning process: {ex.Message}")
+          }
+        | _ -> incorrectArgs ())
+      sqlSpec = NotYetImplemented
+      previewable = Impure
+      deprecated = NotDeprecated }
+
+
+    { name = fn "processGetPid" 0
+      typeParams = []
+      parameters = [ Param.make "unit" TUnit "" ]
+      returnType = TInt64
+      description = "Returns the current process ID (PID)."
+      fn =
+        (function
+        | _, _, _, [ DUnit ] ->
+          uply {
+            let pid = System.Diagnostics.Process.GetCurrentProcess().Id
+            return DInt64(int64 pid)
+          }
+        | _ -> incorrectArgs ())
+      sqlSpec = NotYetImplemented
+      previewable = Impure
+      deprecated = NotDeprecated }
+
+
+    { name = fn "processIsRunning" 0
+      typeParams = []
+      parameters = [ Param.make "pid" TInt64 "Process ID to check" ]
+      returnType = TBool
+      description = "Checks if a process with the given PID is currently running."
+      fn =
+        (function
+        | _, _, _, [ DInt64 pid ] ->
+          uply {
+            try
+              let proc = System.Diagnostics.Process.GetProcessById(int pid)
+              let isRunning = not proc.HasExited
+              return DBool isRunning
+            with
+            | :? System.ArgumentException
+
+            | :? System.InvalidOperationException ->
+              // Process doesn't exist or has exited
+              return DBool false
+          }
+        | _ -> incorrectArgs ())
+      sqlSpec = NotYetImplemented
+      previewable = Impure
+      deprecated = NotDeprecated }
+
+
+    { name = fn "processKill" 0
+      typeParams = []
+      parameters = [ Param.make "pid" TInt64 "Process ID to kill" ]
+      returnType = TypeReference.result TUnit TString
+      description =
+        "Kills the process with the given PID. Returns unit on success, or an error message on failure."
+      fn =
+        (function
+        | _state, _, _, [ DInt64 pid ] ->
+          uply {
+            try
+              let proc = System.Diagnostics.Process.GetProcessById(int pid)
+              proc.Kill()
+              proc.WaitForExit(5000) |> ignore<bool>
+              return Dval.resultOk KTUnit KTString DUnit
+            with
+            | :? System.ArgumentException ->
+              return Dval.resultError KTUnit KTString (DString "Process not found")
+            | ex ->
+              return
+                Dval.resultError
+                  KTUnit
+                  KTString
+                  (DString $"Error killing process: {ex.Message}")
+          }
+        | _ -> incorrectArgs ())
+      sqlSpec = NotYetImplemented
+      previewable = Impure
+      deprecated = NotDeprecated } ]
+
+
+let builtins : Builtins = Builtin.make [] fns

backend/src/BuiltinPM/Libs/PackageOps.fs

@@ -21,32 +21,38 @@ open Builtin.Shortcuts
 let packageOpTypeName =
   FQTypeName.fqPackage PackageIDs.Type.LanguageTools.ProgramTypes.packageOp
 
+let packageOpBatchTypeName =
+  FQTypeName.fqPackage PackageIDs.Type.LanguageTools.ProgramTypes.packageOpBatch
+
 
 // TODO: review/reconsider the accessibility of these fns
 let fns : List<BuiltInFn> =
   [ { name = fn "scmAddOps" 0
       typeParams = []
       parameters =
-        [ Param.make "branchID" (TypeReference.option TUuid) ""
+        [ Param.make "instanceID" (TypeReference.option TUuid) ""
+          Param.make "branchID" (TypeReference.option TUuid) ""
           Param.make "ops" (TList(TCustomType(Ok packageOpTypeName, []))) "" ]
       returnType = TypeReference.result TInt64 TString
       description =
         "Add package ops to the database and apply them to projections.
-        Returns Ok(insertedCount) on success (duplicates are skipped), or Error with message on failure."
+        Pass None for instanceID for local ops, or Some(uuid) for ops from remote instances.
+        Returns the number of inserted ops on success (duplicates are skipped), or an error message on failure."
       fn =
         let resultOk = Dval.resultOk KTInt64 KTString
         let resultError = Dval.resultError KTInt64 KTString
         (function
-        | _, _, _, [ branchID; DList(_vtTODO, ops) ] ->
+        | _, _, _, [ instanceID; branchID; DList(_vtTODO, ops) ] ->
           uply {
             try
               // Deserialize dvals
               let branchID = C2DT.Option.fromDT D.uuid branchID
+              let instanceID = C2DT.Option.fromDT D.uuid instanceID
               let ops = ops |> List.choose PT2DT.PackageOp.fromDT
 
               // Insert ops with deduplication, get count of actually inserted ops
               let! insertedCount =
-                LibPackageManager.Inserts.insertAndApplyOps branchID ops
+                LibPackageManager.Inserts.insertAndApplyOps instanceID branchID ops
 
               return resultOk (DInt64 insertedCount)
             with ex ->
@@ -110,23 +116,49 @@ let fns : List<BuiltInFn> =
     { name = fn "scmGetOpsSince" 0
       typeParams = []
       parameters =
-        [ Param.make "branchID" (TypeReference.option TUuid) ""
+        [ Param.make "targetInstanceID" (TypeReference.option TUuid) ""
           Param.make "since" TDateTime "" ]
-      returnType = TList(TCustomType(Ok packageOpTypeName, []))
-      description = "Get package ops created since the given timestamp."
+      returnType = TList(TCustomType(Ok packageOpBatchTypeName, []))
+      description =
+        "Get all package ops (from ALL branches) created since the given timestamp, grouped by branch and instance.
+        Optionally filters for a target instance (pass None to get all ops, or Some(uuid) to exclude ops from that target instance).
+        Returns a list of PackageOpBatch, where each batch contains ops from one branch with the same instanceID."
       fn =
         function
-        | _, _, _, [ branchID; DDateTime since ] ->
+        | _, _, _, [ targetInstanceID; DDateTime since ] ->
           uply {
-            let branchID = C2DT.Option.fromDT D.uuid branchID
-
-            let! ops = LibPackageManager.Queries.getOpsSince branchID since
-
-            return
-              DList(
-                VT.customType PT2DT.PackageOp.typeName [],
-                ops |> List.map PT2DT.PackageOp.toDT
-              )
+            let targetID = C2DT.Option.fromDT D.uuid targetInstanceID
+
+            let! opsWithMetadata =
+              LibPackageManager.Queries.getAllOpsSince targetID since
+
+            // Group by (branchID, instanceID)
+            let grouped =
+              opsWithMetadata
+              |> List.groupBy (fun (_, branchID, instanceID) ->
+                (branchID, instanceID))
+              |> Map.toList
+
+            // Convert each group to a PackageOpBatch record
+            let batches =
+              grouped
+              |> List.map (fun ((branchID, instanceID), ops) ->
+                let opsList =
+                  ops
+                  |> List.map (fun (op, _, _) -> PT2DT.PackageOp.toDT op)
+                  |> fun opDvals ->
+                      DList(VT.customType packageOpTypeName [], opDvals)
+
+                let fields =
+                  [ ("branchID", branchID |> Option.map DUuid |> Dval.option KTUuid)
+                    ("instanceID",
+                     instanceID |> Option.map DUuid |> Dval.option KTUuid)
+                    ("ops", opsList) ]
+                  |> Map
+
+                DRecord(packageOpBatchTypeName, packageOpBatchTypeName, [], fields))
+
+            return DList(VT.customType packageOpBatchTypeName [], batches)
           }
         | _ -> incorrectArgs ()
       sqlSpec = NotQueryable

backend/src/LibExecution/PackageIDs.fs

@@ -313,6 +313,8 @@ module Type =
           p [] "SearchResults" "0660f9dc-a816-4185-9e5c-f936325f83d5"
 
       let packageOp = p [] "PackageOp" "7d8e9f0a-1b2c-3d4e-5f6a-7b8c9d0e1f2a"
+      let packageOpBatch =
+        p [] "PackageOpBatch" "9f1a2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
 
       let secret = p [] "Secret" "37427120-d71d-41f2-b094-68757570bc41"
       let db = p [] "DB" "7f219668-f8ac-4b17-a404-1171985dadf9"

backend/src/LibPackageManager/Inserts.fs

@@ -38,7 +38,10 @@ let computeOpHash (op : PT.PackageOp) : System.Guid =
 
 /// Insert PackageOps into the package_ops table and apply them to projection tables
 /// Returns the count of ops actually inserted (duplicates are skipped via INSERT OR IGNORE)
+/// CLEANUP: The 'applied' flag is currently always set to true and all ops are applied
+/// immediately. Should we reconsider this?
 let insertAndApplyOps
+  (instanceID : Option<System.Guid>)
   (branchID : Option<PT.BranchID>)
   (ops : List<PT.PackageOp>)
   : Task<int64> =
@@ -60,8 +63,8 @@ let insertAndApplyOps
 
           let sql =
             """
-            INSERT OR IGNORE INTO package_ops (id, branch_id, op_blob, applied)
-            VALUES (@id, @branch_id, @op_blob, @applied)
+            INSERT OR IGNORE INTO package_ops (id, branch_id, op_blob, applied, instance_id)
+            VALUES (@id, @branch_id, @op_blob, @applied, @instance_id)
             """
 
           let parameters =
@@ -71,7 +74,11 @@ let insertAndApplyOps
                | Some id -> Sql.uuid id
                | None -> Sql.dbnull)
               "op_blob", Sql.bytes opBlob
-              "applied", Sql.bool true ]
+              "applied", Sql.bool true
+              "instance_id",
+              (match instanceID with
+               | Some id -> Sql.uuid id
+               | None -> Sql.dbnull) ]
 
           (sql, [ parameters ]))