Unify log observers into single logObserver field, auto-disable ANSI in agent mode

- Combine ansiLogObserver/agentLogObserver into single LogObserver interface + logObserver field on Session (only one active at a time)
- Auto-disable ANSI log in CliOptions when agent mode detected, so task hashes and submission lines flow through ConsoleAppender
- Both AnsiLogObserver and AgentLogObserver implement LogObserver
- LoggerHelper uses instanceof dispatch for ANSI-specific features (appendSticky, started/stopped checks)

Signed-off-by: Edmund Miller <edmund.miller@seqera.io>

Co-authored-by: Ben Sherman <bentshermann@gmail.com>
Co-authored-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

Author: 3 people

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -70,7 +70,7 @@ import nextflow.script.WorkflowMetadata
 import nextflow.script.dsl.ProcessConfigBuilder
 import nextflow.spack.SpackConfig
 import nextflow.trace.AgentLogObserver
-import nextflow.trace.AnsiLogObserver
+import nextflow.trace.LogObserver
 import nextflow.trace.TraceObserver
 import nextflow.trace.TraceObserverFactory
 import nextflow.trace.TraceObserverFactoryV2
@@ -317,9 +317,7 @@ class Session implements ISession {
 
     boolean disableJobsCancellation
 
-    AnsiLogObserver ansiLogObserver
-
-    AgentLogObserver agentLogObserver
+    LogObserver logObserver
 
     FilePorter getFilePorter() { filePorter }
 
@@ -506,9 +504,10 @@ class Session implements ISession {
         result.add(statsObserver)
         // Add AgentLogObserver when agent mode is enabled
         if( agentLog ) {
-            this.agentLogObserver = new AgentLogObserver()
-            agentLogObserver.setStatsObserver(statsObserver)
-            result.add(agentLogObserver)
+            def observer = new AgentLogObserver()
+            observer.setStatsObserver(statsObserver)
+            this.logObserver = observer
+            result.add(observer)
         }
         for( TraceObserverFactoryV2 f : Plugins.getExtensions(TraceObserverFactoryV2) ) {
             log.debug "Observer factory (v2): ${f.class.simpleName}"
@@ -844,8 +843,7 @@ class Session implements ISession {
             shutdown0()
             notifyError(null)
             // force termination
-            ansiLogObserver?.forceTermination()
-            agentLogObserver?.forceTermination()
+            logObserver?.forceTermination()
             executorFactory?.signalExecutors()
             processesBarrier.forceTermination()
             monitorsBarrier.forceTermination()
@@ -1295,16 +1293,16 @@ class Session implements ISession {
     }
 
     void printConsole(String str, boolean newLine=false) {
-        if( ansiLogObserver )
-            ansiLogObserver.appendInfo(str)
+        if( logObserver )
+            logObserver.appendInfo(str)
         else if( newLine )
             System.out.println(str)
         else
             System.out.print(str)
     }
 
     void printConsole(Path file) {
-        ansiLogObserver ? ansiLogObserver.appendInfo(file.text) : Files.copy(file, System.out)
+        logObserver ? logObserver.appendInfo(file.text) : Files.copy(file, System.out)
     }
 
     private volatile ThreadPoolManager finalizePoolManager

modules/nextflow/src/main/groovy/nextflow/cli/CliOptions.groovy

@@ -118,6 +118,12 @@ class CliOptions {
         if( noColor ) {
             return ansiLog = false
         }
+
+        // Disable ANSI log in agent mode for plain, parseable output
+        if( SysEnv.isAgentMode() ) {
+            return ansiLog = false
+        }
+
         return Ansi.isEnabled()
     }
 

modules/nextflow/src/main/groovy/nextflow/trace/AgentLogObserver.groovy

@@ -39,7 +39,7 @@ import nextflow.trace.event.TaskEvent
  * @author Edmund Miller <edmund.miller@utdallas.edu>
  */
 @Slf4j
-class AgentLogObserver implements TraceObserverV2 {
+class AgentLogObserver implements TraceObserverV2, LogObserver {
 
     private Session session
     private WorkflowStatsObserver statsObserver

modules/nextflow/src/main/groovy/nextflow/trace/AnsiLogObserver.groovy

@@ -39,7 +39,7 @@ import org.fusesource.jansi.AnsiConsole
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
 @CompileStatic
-class AnsiLogObserver implements TraceObserverV2 {
+class AnsiLogObserver implements TraceObserverV2, LogObserver {
 
     static final private String NEWLINE = '\n'
 

modules/nextflow/src/main/groovy/nextflow/trace/DefaultObserverFactory.groovy

@@ -48,8 +48,9 @@ class DefaultObserverFactory implements TraceObserverFactoryV2 {
 
     protected void createAnsiLogObserver(Collection<TraceObserverV2> result) {
         if( session.ansiLog ) {
-            session.ansiLogObserver = new AnsiLogObserver()
-            result << session.ansiLogObserver
+            def observer = new AnsiLogObserver()
+            session.logObserver = observer
+            result << observer
         }
     }
 

modules/nextflow/src/main/groovy/nextflow/trace/LogObserver.groovy

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2013-2024, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package nextflow.trace
+
+/**
+ * Common interface for log observer implementations (ANSI and Agent modes).
+ *
+ * Only one log observer is active at a time. This interface defines the
+ * shared contract used by {@link nextflow.Session} and
+ * {@link nextflow.util.LoggerHelper}.
+ *
+ * @author Edmund Miller <edmund.miller@utdallas.edu>
+ */
+interface LogObserver {
+
+    void appendInfo(String message)
+
+    void appendWarning(String message)
+
+    void appendError(String message)
+
+    void forceTermination()
+}

modules/nextflow/src/main/groovy/nextflow/util/LoggerHelper.groovy

@@ -56,6 +56,7 @@ import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowWriteChannel
 import nextflow.Global
 import nextflow.Session
+import nextflow.trace.AnsiLogObserver
 import nextflow.cli.CliOptions
 import nextflow.cli.Launcher
 import nextflow.exception.AbortOperationException
@@ -714,33 +715,31 @@ class LoggerHelper {
             try {
                 final message = fmtEvent(event, session, false)
 
-                // Forward to AgentLogObserver when agent mode is enabled
-                final agentRenderer = session?.agentLogObserver
-                if( agentRenderer ) {
+                final observer = session?.logObserver
+                if( observer instanceof AnsiLogObserver ) {
+                    final renderer = (AnsiLogObserver)observer
+                    if( !renderer.started || renderer.stopped )
+                        System.out.println(message)
+                    else if( event.marker == STICKY )
+                        renderer.appendSticky(message)
+                    else if( event.level==Level.ERROR )
+                        renderer.appendError(message)
+                    else if( event.level==Level.WARN )
+                        renderer.appendWarning(message)
+                    else
+                        renderer.appendInfo(message)
+                }
+                else if( observer ) {
                     if( event.level==Level.ERROR )
-                        agentRenderer.appendError(message)
+                        observer.appendError(message)
                     else if( event.level==Level.WARN )
-                        agentRenderer.appendWarning(message)
+                        observer.appendWarning(message)
                     else
-                        agentRenderer.appendInfo(message)
-                    return
+                        observer.appendInfo(message)
                 }
-
-                final renderer = session?.ansiLogObserver
-                if( !renderer || !renderer.started || renderer.stopped )
+                else {
                     System.out.println(message)
-
-                else if( event.marker == STICKY )
-                    renderer.appendSticky(message)
-
-                else if( event.level==Level.ERROR )
-                    renderer.appendError(message)
-
-                else if( event.level==Level.WARN )
-                    renderer.appendWarning(message)
-
-                else
-                    renderer.appendInfo(message)
+                }
             }
             catch (Throwable e) {
                 e.printStackTrace()