/****************************************************************************
* Copyright (c) 2022-2025 Imixs Software Solutions GmbH and others.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v. 2.0 which is available at
* https://www.eclipse.org/legal/epl-2.0
*
* This Source Code may also be made available under the terms of the
* GNU General Public License, version 2 or later (GPL-2.0-or-later),
* which is available at https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
*
* SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
****************************************************************************/
package ai.dynamixs.cognition;
import java.util.List;
import java.util.Optional;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.imixs.ai.ImixsAIContextHandler;
import org.imixs.ai.workflow.OpenAIAPIConnector;
import org.imixs.ai.workflow.OpenAIAPIService;
import org.imixs.workflow.ItemCollection;
import org.imixs.workflow.engine.AsyncEventScheduler;
import org.imixs.workflow.engine.EventLogService;
import org.imixs.workflow.engine.WorkflowService;
import org.imixs.workflow.engine.jpa.EventLog;
import org.imixs.workflow.exceptions.AdapterException;
import org.imixs.workflow.exceptions.PluginException;
import jakarta.annotation.security.DeclareRoles;
import jakarta.annotation.security.RunAs;
import jakarta.ejb.LocalBean;
import jakarta.ejb.Stateless;
import jakarta.ejb.TransactionAttribute;
import jakarta.ejb.TransactionAttributeType;
import jakarta.inject.Inject;
import jakarta.persistence.OptimisticLockException;
/**
* The CognitoEventService is responsible to process Dynamixs-AI Cognito events
* in an asynchronous batch process. The CognitoEventService automatically
* lookup eventLog entries of the topic "rag.event.*". The RAGEventService is
* called only by the RAGEventScheduler which is implementing a
* ManagedScheduledExecutorService.
*
* The CognitoEventService reacts on events from the following types:
*
* - cognito.event.prompt - resoles the llm prompt based on the given workitem
* and creates a cache event
* - cognito.event.cache - indicates that this entry is still cached in the
* event log database
*
* The processor look up the workItem by the $uniqueId.
*
* To prevent concurrent processes to handle the same workitems the batch
* process uses a Optimistic lock strategy. After fetching new event log entries
* the processor updates the eventLog entry in a new transaction and set the
* topic to 'batch.process.lock'. After that update we can be sure that no other
* process is dealing with these entries. After completing the processing step
* the eventlog entry will be removed.
*
* To avoid ad deadlock the processor set an expiration time on the lock, so the
* lock will be auto-released after 1 minute (batch.processor.deadlock).
*
* @see AsyncEventScheduler
* @version 1.0
* @author rsoika
*
*/
@DeclareRoles({ "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
@RunAs("org.imixs.ACCESSLEVEL.MANAGERACCESS")
@Stateless
@LocalBean
public class CognitoEventService {
public static final String EVENTLOG_TOPIC_COGNITO_PROMPT = "cognito.event.prompt";
public static final String EVENTLOG_TOPIC_COGNITO_CACHE = "cognito.event.cache";
public static final String ITEM_PROMPT_DEFINITION = "prompt.definition";
private static final Logger logger = Logger.getLogger(CognitoEventService.class.getName());
@Inject
EventLogService eventLogService;
@Inject
private WorkflowService workflowService;
@Inject
private OpenAIAPIService llmService;
@Inject
ImixsAIContextHandler imixsAIContextHandler;
@Inject
CognitoLockHandler cognitoLockHandler;
@Inject
@ConfigProperty(name = OpenAIAPIConnector.ENV_LLM_SERVICE_ENDPOINT)
Optional serviceEndpoint;
/**
* The method lookups for cognito prompt event log entries and updates the ....
* information
*
* Each eventLogEntry provides optional a prompt template
*
*/
@TransactionAttribute(value = TransactionAttributeType.REQUIRES_NEW)
public void processEventLog() {
long l = System.currentTimeMillis();
// test for new event log entries by timeout...
List events = eventLogService.findEventsByTimeout(100,
CognitoEventService.EVENTLOG_TOPIC_COGNITO_PROMPT, CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE);
if (events.size() > 0) {
logger.log(Level.INFO, "├── 🔃 processing {0} Cognito events....", events.size());
// do we know the endpoint?
if (!serviceEndpoint.isPresent()) {
logger.log(Level.WARNING,
"├── ⚠️ Missing environment var 'llm.service.endpoint' - can not process cognito events!");
return;
}
for (EventLog eventLogEntry : events) {
try {
boolean llmAPIDebug = true;
logger.log(Level.INFO,
"│ ├── Event: " + eventLogEntry.getTopic() + " - " + eventLogEntry.getRef());
// first try to lock the eventLog entry....
if (eventLogService.lock(eventLogEntry)) {
// Index / Update event?
ItemCollection workitem = workflowService.getWorkItem(eventLogEntry.getRef());
if (workitem == null) {
logger.log(Level.WARNING, "│ ├── ⚠️ unable to read workitem: " + eventLogEntry.getRef());
eventLogService.removeEvent(eventLogEntry.getId());
continue;
}
ItemCollection promptItemCol = new ItemCollection(eventLogEntry.getData());
switch (eventLogEntry.getTopic()) {
case CognitoEventService.EVENTLOG_TOPIC_COGNITO_PROMPT + ".lock":
// load the promptDefinition from the event data object
String promptTemplate = promptItemCol.getItemValueString(ITEM_PROMPT_DEFINITION);
imixsAIContextHandler.setWorkItem(workitem);
imixsAIContextHandler.addPromptDefinition(promptTemplate);
String llmPrompt = imixsAIContextHandler.toString();
// if we have a prompt we call the llm api endpoint
if (llmPrompt != null && !llmPrompt.isBlank()) {
String llmAPIEndpoint = serviceEndpoint.get();
// postPromptCompletion
// JsonObject jsonPrompt =
// llmService.buildJsonPromptObjectV1(imixsAIContextHandler);
String completionResult = llmService.postPromptCompletion(imixsAIContextHandler,
llmAPIEndpoint);
// process the ai.result....
if (llmAPIDebug) {
logger.info("===> Completion Result: ");
logger.info(completionResult);
}
String resultMessage = llmService.processPromptResult(completionResult, "", workitem);
// if the workitem is not locked we can safety update the $workflowabstarct
if (!cognitoLockHandler.isLocked(workitem.getUniqueID())) {
logger.log(Level.INFO,
"│ ├── ✓ workitem not locked - updating $workflowabstract...");
workitem.setItemValue("$workflowabstract", resultMessage);
workflowService.getDocumentService().save(workitem);
} else {
// workitem is still in edit mode - so we create a cache event that just stores
// the result. This is an async caching mechanism for the front end.
logger.log(Level.INFO,
"│ ├── ⚐ workitem locked - create COGNITO_CACHE event...");
promptItemCol.setItemValue("$workflowabstract", resultMessage);
// create a new event
eventLogService.createEvent(CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE,
workitem.getUniqueID(),
promptItemCol);
}
} else {
logger.finest(
"......no prompt definition found for " + workitem.getUniqueID());
}
// finally remove the event log entry...
eventLogService.removeEvent(eventLogEntry.getId());
break;
case CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE + ".lock":
logger.log(Level.INFO,
"│ ├── process COGNITO_CACHE event...");
if (!cognitoLockHandler.isLocked(workitem.getUniqueID())) {
logger.log(Level.INFO,
"│ ├── ✓ workitem not locked - updating $workflowabstract...");
// ItemCollection promptItemCol = new ItemCollection(eventLogEntry.getData());
String resultMessage = promptItemCol.getItemValueString("$workflowabstract");
workitem.setItemValue("$workflowabstract", resultMessage);
workflowService.getDocumentService().save(workitem);
// finally remove the event log entry...
eventLogService.removeEvent(eventLogEntry.getId());
} else {
// repeat event until workitem is unlocked
eventLogService.unlock((eventLogEntry));
}
break;
default:
// no op
}
}
} catch (OptimisticLockException e) {
// lock was not possible - continue....
logger.log(Level.INFO, "│ ├── ⚠️ unable to lock AsyncEvent: {0}", e.getMessage());
} catch (PluginException | AdapterException e) {
logger.log(Level.WARNING, e.getErrorCode() + " - " + e.getMessage());
// remove the event log entry...
eventLogService.removeEvent(eventLogEntry.getId());
}
}
logger.log(Level.INFO, "├── ✅ {0} CognitoEvents processed in {1}ms",
new Object[] { events.size(), System.currentTimeMillis() - l });
}
}
}