diff options
Diffstat (limited to 'Omni/Agent/Worker.hs')
| -rw-r--r-- | Omni/Agent/Worker.hs | 75 |
1 files changed, 20 insertions, 55 deletions
diff --git a/Omni/Agent/Worker.hs b/Omni/Agent/Worker.hs index 94a4e35..c01a853 100644 --- a/Omni/Agent/Worker.hs +++ b/Omni/Agent/Worker.hs @@ -5,11 +5,14 @@ module Omni.Agent.Worker where import Alpha +import Control.Concurrent (forkIO, killThread, threadDelay) +import Control.Monad (forever) import qualified Data.Aeson as Aeson import qualified Data.Aeson.KeyMap as KM import qualified Data.ByteString.Lazy as BL import qualified Data.Scientific as Scientific import qualified Data.Text as Text +import qualified Data.Text.IO as TIO import qualified Data.Time as Time import qualified Omni.Agent.Core as Core import qualified Omni.Agent.Git as Git @@ -152,7 +155,10 @@ runAmp repo task = do <> "'.\n" Directory.createDirectoryIfMissing True (repo </> "_/llm") - let logFile = repo </> "_/llm/amp.log" + let logPath = repo </> "_/llm/amp.log" + + -- Ensure log file is empty/exists + IO.writeFile logPath "" -- Read AGENTS.md agentsMd <- @@ -167,13 +173,8 @@ runAmp repo task = do <> "\n\nREPOSITORY GUIDELINES (AGENTS.md):\n" <> agentsMd - -- Clean up previous log - exists <- Directory.doesFileExist logFile - when exists (Directory.removeFile logFile) - - -- Start background monitors - tidTime <- forkIO timeTicker - tidLog <- forkIO (monitorLog logFile) + -- Monitor log file + tidLog <- forkIO (monitorLog logPath) -- Assume amp is in PATH let args = ["--log-level", "debug", "--log-file", "_/llm/amp.log", "--dangerously-allow-all", "-x", Text.unpack fullPrompt] @@ -182,7 +183,6 @@ runAmp repo task = do (exitCode, out, _err) <- Process.readCreateProcessWithExitCode cp "" -- Cleanup - killThread tidTime killThread tidLog pure (exitCode, Text.pack out) @@ -239,56 +239,21 @@ findBaseBranch repo task = do [] -> pure "live" monitorLog :: FilePath -> IO () -monitorLog logPath = do - waitForFile logPath - IO.withFile logPath IO.ReadMode <| \h -> do - -- Start from beginning of file (don't seek to end) +monitorLog path = do + -- Wait for file to exist + waitForFile path + + IO.withFile path IO.ReadMode <| \h -> do + IO.hSetBuffering h IO.LineBuffering forever <| do eof <- IO.hIsEOF h if eof then threadDelay 100000 -- 0.1s else do - line <- IO.hGetLine h - parseAndUpdate (Text.pack line) + line <- TIO.hGetLine h + AgentLog.processLogLine line waitForFile :: FilePath -> IO () -waitForFile path = do - exists <- Directory.doesFileExist path - if exists - then pure () - else do - threadDelay 100000 - waitForFile path - -parseAndUpdate :: Text -> IO () -parseAndUpdate line = do - let maybeObj = Aeson.decode (BL.fromStrict (encodeUtf8 line)) :: Maybe Aeson.Object - case maybeObj of - Nothing -> pure () - Just obj -> do - -- Extract message (was msg) - case KM.lookup "message" obj of - Just (Aeson.String m) -> unless (Text.null m) (AgentLog.updateActivity m) - _ -> pure () - - -- Extract threadId - case KM.lookup "threadId" obj of - Just (Aeson.String tid) -> AgentLog.update (\s -> s {AgentLog.statusThreadId = Just tid}) - _ -> pure () - - -- Extract cost from usage-ledger:event - -- Pattern: {"totalCredits": 154.0, "message": "usage-ledger:event", ...} - -- The credits are in cents, so we divide by 100 to get dollars. - case KM.lookup "totalCredits" obj of - Just (Aeson.Number n) -> - let total = Scientific.toRealFloat n / 100.0 - in AgentLog.update (\s -> s {AgentLog.statusCredits = total}) - _ -> pure () - -timeTicker :: IO () -timeTicker = - forever <| do - time <- Time.getCurrentTime - let timeStr = Time.formatTime Time.defaultTimeLocale "%H:%M" time - AgentLog.update (\s -> s {AgentLog.statusTime = Text.pack timeStr}) - threadDelay 1000000 -- 1s +waitForFile p = do + e <- Directory.doesFileExist p + if e then pure () else threadDelay 100000 >> waitForFile p |
