+ * The Scheduler runs on a non-persistent ejb timer with the same interval + * settings as the AsyncEventProcessor + *
+ * Configuration is read from the environment variables + * 'ASYNCEVENT_PROCESSOR_INTERVAL' and an optional delay defined by + * 'ASYNCEVENT_PROCESSOR_INITIALDELAY'. To enable the processor + * 'ASYNCEVENT_PROCESSOR_ENABLED' must be set to true (default=false). + * 'ASYNCEVENT_PROCESSOR_DEADLOCK' deadlock timeout + *
+ * In a clustered environment this timer runs in each cluster member that + * contains the EJB. So this means the non-persistent EJB Timer scales + * horizontal within a clustered environment – e.g. a Kubernetes cluster. + * + * @see AsyncEventService + * @version 1.1 + * @author rsoika + * + */ +@DeclareRoles({ "org.imixs.ACCESSLEVEL.MANAGERACCESS" }) +@RunAs("org.imixs.ACCESSLEVEL.MANAGERACCESS") +@Startup +@Singleton +public class CognitoEventScheduler { + + // enabled + @Inject + @ConfigProperty(name = ASYNCEVENT_PROCESSOR_ENABLED, defaultValue = "false") + boolean enabled; + + // timeout interval in ms + @Inject + @ConfigProperty(name = ASYNCEVENT_PROCESSOR_INTERVAL, defaultValue = "1000") + long interval; + + // initial delay in ms + @Inject + @ConfigProperty(name = ASYNCEVENT_PROCESSOR_INITIALDELAY, defaultValue = "0") + long initialDelay; + + // deadlock timeout interval in ms + @Inject + @ConfigProperty(name = ASYNCEVENT_PROCESSOR_DEADLOCK, defaultValue = "60000") + long deadLockInterval; + + private static final Logger logger = Logger.getLogger(CognitoEventScheduler.class.getName()); + + @Resource + TimerService timerService; + + @Inject + CognitoEventService cognitoEventService; + + @Inject + EventLogService eventLogService; + + @PostConstruct + public void init() { + if (enabled) { + + // Registering a non-persistent Timer Service. + final TimerConfig timerConfig = new TimerConfig(); + timerConfig.setInfo("Dynamixs AI CognitoEventScheduler"); + timerConfig.setPersistent(false); + timerService.createIntervalTimer(initialDelay, interval, timerConfig); + logger.log(Level.INFO, "├── ✅ Started CognitoEventScheduler - initalDelay={0} inverval={1}", + new Object[] { initialDelay, interval }); + } + } + + /** + * The method delegates the event processing to the stateless ejb + * AsyncEventProcessor. + *
+ * Before processing the eventLog the method releases possible dead locks first. + * Both methods are running in separate transactions + * + */ + @Timeout + public void run(Timer timer) { + eventLogService.releaseDeadLocks(deadLockInterval, + CognitoEventService.EVENTLOG_TOPIC_COGNITO_PROMPT, + CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE); + cognitoEventService.processEventLog(); + } + +} diff --git a/dynamixs-ai-app/src/main/java/ai/dynamixs/cognition/CognitoEventService.java b/dynamixs-ai-app/src/main/java/ai/dynamixs/cognition/CognitoEventService.java new file mode 100644 index 0000000..2117337 --- /dev/null +++ b/dynamixs-ai-app/src/main/java/ai/dynamixs/cognition/CognitoEventService.java @@ -0,0 +1,243 @@ +/**************************************************************************** + * 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.json.JsonObject; +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: + *
+ * 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
+ * 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
+ * The IDs are stored in the lock cache. The CognitoEventService looks up an ID
+ * in this cache before updating a workitem in a safe way.
+ *
+ * It may happen that a user switches into edit mode
+ * ({@link WorkflowEvent#DOCUMENT_CHANGED}) without updating the workitem. For
+ * this reason a locked ID will be released after {@value #LOCK_TIMEOUT_SECONDS}
+ * seconds automatically.
+ *
+ * Thread Safety: This implementation uses a
+ * {@link ConcurrentHashMap} to ensure thread-safe operations in a
+ * multi-user-environment where the same workitem can be opened by different
+ * users in parallel.
+ *
+ * @author rsoika
+ * @version 1.0
+ */
+@ApplicationScoped
+public class CognitoLockHandler {
+
+ private static final Logger logger = Logger.getLogger(CognitoLockHandler.class.getName());
+
+ /**
+ * Lock timeout in seconds. After this duration, a lock is considered stale and
+ * will be automatically released.
+ */
+ public static final long LOCK_TIMEOUT_SECONDS = 120;
+
+ /**
+ * Thread-safe cache storing workitem uniqueIds mapped to their lock timestamp.
+ * The timestamp is used to determine if a lock has expired.
+ */
+ private final Map
+ *
+ * This observer is executed with {@link Interceptor.Priority#LIBRARY_BEFORE}
+ * priority to ensure it runs before other observers.
+ *
+ * @param workflowEvent the workflow event containing the workitem and event
+ * type
+ * @throws PluginException if an error occurs during event processing
+ */
+ public void onWorkflowEvent(@Observes WorkflowEvent workflowEvent)
+ throws PluginException {
+
+ int eventType = workflowEvent.getEventType();
+ String uniqueId = extractUniqueId(workflowEvent);
+
+ if (uniqueId == null || uniqueId.isEmpty()) {
+ logger.log(Level.FINE, "Skipping lock operation - workitem has no uniqueId");
+ return;
+ }
+
+ // Lock the workitem when user enters edit mode
+ if (WorkflowEvent.WORKITEM_CHANGED == eventType) {
+ lock(uniqueId);
+ logger.log(Level.INFO, "Workitem locked: uniqueId={0}, active locks={1}",
+ new Object[] { uniqueId, lockCache.size() });
+ }
+
+ // Unlock the workitem after successful save
+ if (WorkflowEvent.WORKITEM_AFTER_PROCESS == eventType) {
+ boolean wasLocked = unlock(uniqueId);
+ if (wasLocked) {
+ logger.log(Level.INFO, "Workitem unlocked after process: uniqueId={0}, remaining locks={1}",
+ new Object[] { uniqueId, lockCache.size() });
+ } else {
+ logger.log(Level.INFO, "Workitem was not locked (possibly expired): uniqueId={0}",
+ uniqueId);
+ }
+ }
+ }
+
+ /**
+ * Checks if the given workitem uniqueID is currently locked (in edit mode).
+ *
+ * This method also performs cleanup of stale locks. If the lock for the given
+ * ID has exceeded the timeout ({@value #LOCK_TIMEOUT_SECONDS} seconds), it will
+ * be automatically removed and this method returns {@code false}.
+ *
+ * @param id the unique identifier of the workitem to check
+ * @return {@code true} if the workitem is currently locked and the lock has not
+ * expired, {@code false} otherwise
+ */
+ public boolean isLocked(String id) {
+ if (id == null || id.isEmpty()) {
+ logger.log(Level.FINE, "isLocked called with null or empty id - returning false");
+ return false;
+ }
+
+ Instant lockTimestamp = lockCache.get(id);
+
+ if (lockTimestamp == null) {
+ logger.log(Level.FINE, "Workitem not locked: uniqueId={0}", id);
+ return false;
+ }
+
+ // Check if lock has expired
+ if (isLockExpired(lockTimestamp)) {
+ logger.log(Level.INFO, "Lock expired and removed: uniqueId={0}, locked since={1}",
+ new Object[] { id, lockTimestamp });
+ lockCache.remove(id);
+ return false;
+ }
+
+ long lockedSeconds = java.time.Duration.between(lockTimestamp, Instant.now()).getSeconds();
+ logger.log(Level.FINE, "Workitem is locked: uniqueId={0}, locked for {1}s",
+ new Object[] { id, lockedSeconds });
+ return true;
+ }
+
+ /**
+ * Adds a workitem uniqueID to the lock cache with the current timestamp.
+ *
+ * If the ID is already in the cache, the timestamp will be updated (lock
+ * refresh).
+ *
+ * @param id the unique identifier of the workitem to lock
+ */
+ public void lock(String id) {
+ if (id == null || id.isEmpty()) {
+ logger.log(Level.WARNING, "Attempted to lock with null or empty id - operation ignored");
+ return;
+ }
+
+ Instant previousLock = lockCache.put(id, Instant.now());
+
+ if (previousLock != null) {
+ logger.log(Level.FINE, "Lock refreshed: uniqueId={0}, previous lock from={1}",
+ new Object[] { id, previousLock });
+ } else {
+ logger.log(Level.FINE, "New lock created: uniqueId={0}", id);
+ }
+
+ // Opportunistic cleanup of expired locks
+ cleanupExpiredLocks();
+ }
+
+ /**
+ * Removes a workitem uniqueID from the lock cache.
+ *
+ * @param id the unique identifier of the workitem to unlock
+ * @return {@code true} if the workitem was locked and is now unlocked,
+ * {@code false} if the workitem was not locked
+ */
+ public boolean unlock(String id) {
+ if (id == null || id.isEmpty()) {
+ logger.log(Level.WARNING, "Attempted to unlock with null or empty id - operation ignored");
+ return false;
+ }
+
+ Instant removed = lockCache.remove(id);
+
+ if (removed != null) {
+ long lockedDuration = java.time.Duration.between(removed, Instant.now()).getSeconds();
+ logger.log(Level.FINE, "Lock removed: uniqueId={0}, was locked for {1}s",
+ new Object[] { id, lockedDuration });
+ return true;
+ }
+
+ return false;
+ }
+
+ /**
+ * Returns the current number of active locks in the cache.
+ *
+ * Note: This count may include expired locks that have not yet
+ * been cleaned up.
+ *
+ * @return the number of entries in the lock cache
+ */
+ public int getLockCount() {
+ return lockCache.size();
+ }
+
+ /**
+ * Removes all expired locks from the cache.
+ *
+ * This method is called opportunistically during lock operations to prevent
+ * unbounded cache growth.
+ */
+ public void cleanupExpiredLocks() {
+ int initialSize = lockCache.size();
+
+ lockCache.entrySet().removeIf(entry -> {
+ boolean expired = isLockExpired(entry.getValue());
+ if (expired) {
+ logger.log(Level.FINE, "Cleaning up expired lock: uniqueId={0}, locked since={1}",
+ new Object[] { entry.getKey(), entry.getValue() });
+ }
+ return expired;
+ });
+
+ int removed = initialSize - lockCache.size();
+ if (removed > 0) {
+ logger.log(Level.INFO, "Expired locks cleanup completed: removed={0}, remaining={1}",
+ new Object[] { removed, lockCache.size() });
+ }
+ }
+
+ /**
+ * Checks if a lock timestamp has exceeded the timeout threshold.
+ *
+ * @param lockTimestamp the timestamp when the lock was created
+ * @return {@code true} if the lock is expired, {@code false} otherwise
+ */
+ private boolean isLockExpired(Instant lockTimestamp) {
+ return lockTimestamp.plusSeconds(LOCK_TIMEOUT_SECONDS).isBefore(Instant.now());
+ }
+
+ /**
+ * Extracts the uniqueId from the workitem contained in the workflow event.
+ *
+ * @param workflowEvent the workflow event
+ * @return the uniqueId of the workitem, or {@code null} if not available
+ */
+ private String extractUniqueId(WorkflowEvent workflowEvent) {
+ if (workflowEvent == null || workflowEvent.getWorkitem() == null) {
+ logger.log(Level.WARNING, "WorkflowEvent or workitem is null - cannot extract uniqueId");
+ return null;
+ }
+
+ String uniqueId = workflowEvent.getWorkitem().getUniqueID();
+
+ if (uniqueId == null || uniqueId.isEmpty()) {
+ logger.log(Level.FINE, "Workitem has no uniqueId (possibly a new document)");
+ }
+
+ return uniqueId;
+ }
+}
\ No newline at end of file
diff --git a/dynamixs-ai-app/src/main/java/ai/dynamixs/cognition/CognitoPlugin.java b/dynamixs-ai-app/src/main/java/ai/dynamixs/cognition/CognitoPlugin.java
new file mode 100644
index 0000000..f6612c7
--- /dev/null
+++ b/dynamixs-ai-app/src/main/java/ai/dynamixs/cognition/CognitoPlugin.java
@@ -0,0 +1,137 @@
+/****************************************************************************
+ * 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.logging.Logger;
+
+import org.imixs.ai.workflow.ImixsAIPromptService;
+import org.imixs.workflow.ItemCollection;
+import org.imixs.workflow.engine.EventLogService;
+import org.imixs.workflow.engine.WorkflowService;
+import org.imixs.workflow.exceptions.PluginException;
+
+import ai.dynamixs.ChronicleTextBuilder;
+import jakarta.inject.Inject;
+
+/**
+ * The CognitoPlugin is a generic plugin to generate an cognito.create.event.
+ * This event is processed by the `CognitoEventService` to generate an AI based
+ * workflow abstract
+ *
+ * The plugin can be activated by the following workflow result tag
+ *
+ *
+ * The embedded tag 'debug' is optional
+ *
+ *
+ * The prompt definition is either loaded from the corresponding BPMN event
+ * (DataObject) or generated as generic default template in case that no
+ * PromptDefinition is assigned to the current event.
+ *
+ *
+ * @author rsoika
+ *
+ */
+public class CognitoPlugin extends org.imixs.ai.rag.RAGIndexPlugin {
+
+ private static final Logger logger = Logger.getLogger(CognitoPlugin.class.getName());
+
+ @Inject
+ private WorkflowService workflowService;
+
+ @Inject
+ private EventLogService eventLogService;
+
+ @Inject
+ private ImixsAIPromptService imixsAIPromptService;
+
+ @Override
+ public ItemCollection run(ItemCollection workitem, ItemCollection event) throws PluginException {
+
+ // do only run on type = workitem!
+ if (!workitem.getType().equals("workitem")) {
+ // no op!
+ return workitem;
+ }
+
+ // read optional configuration form the model or imixs.properties....
+ List
+ *
+ *
+ {@code
+
+ *
+
+
+
+
+
+
+
-E-Mail:
-Address:
-
-
+