luigidemasi commented on code in PR #27649: URL: https://github.com/apache/camel/pull/27649#discussion_r4237165865
########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/MemorySemanticAuditStore.java: ########## @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.semantic; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +import org.apache.camel.support.service.ServiceSupport; + +/** Bounded recent history, lost when the context is destroyed. No persistent-storage guarantee. */ +public final class MemorySemanticAuditStore extends ServiceSupport implements SemanticAuditSink, SemanticAuditReader { + private final int capacity; + private final String epoch = UUID.randomUUID().toString(); + private final Deque<Entry> records = new ArrayDeque<>(); + private long sequence; + private long evicted; + + public MemorySemanticAuditStore(int capacity) { + if (capacity < 1 || capacity > 100000) { + throw new IllegalArgumentException("Invalid audit capacity"); + } + this.capacity = capacity; + } + + @Override + public synchronized void append(SemanticAuditRecord record) { + if (records.size() == capacity) { + records.removeLast(); + evicted++; + } + records.addFirst(new Entry(++sequence, record)); + } + + @Override + public synchronized Optional<SemanticAuditRecord> get(String eventId) { + return records.stream().map(Entry::record).filter(r -> r.getEventId().equals(eventId)).findFirst(); + } + + @Override + public synchronized Map<String, SemanticAuditRecord> getAll(List<String> eventIds) { + if (eventIds.size() > 100) { + throw new IllegalArgumentException("Too many audit evidence references"); + } + var selected = new HashSet<>(eventIds); + Map<String, SemanticAuditRecord> found = new LinkedHashMap<>(); + for (Entry entry : records) { + if (selected.remove(entry.record.getEventId())) { + found.put(entry.record.getEventId(), entry.record); + } + } + return Map.copyOf(found); + } + + @Override + public synchronized SemanticAuditPage query(SemanticAuditQuery query) { + long before = Long.MAX_VALUE; + if (query.getCursor() != null) { + String[] parts = query.getCursor().split(":", -1); + if (parts.length != 3 || !epoch.equals(parts[0]) + || !Integer.toString(query.getFilters().hashCode() + Objects.hashCode(query.getSince())).equals(parts[2])) { Review Comment: Fixed in `fa99d7d1407991cf4c2e0bafbf9ce1e6f146a1aa` using a SHA-256 fingerprint of canonical, sorted filter names/values and the normalized since timestamp. Objects.hash would still collide: expert=Aa and expert=BB produce the same hash with both formulas. The regression now rejects that wrong-filter cursor, rejects a changed since value, and accepts equivalent filters supplied in a different order. This binds paging to its query; it is not an authentication mechanism. _AI-generated by OpenAI Codex on behalf of @luigidemasi via /oss-address-review._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/MemorySemanticAuditStore.java: ########## @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.semantic; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +import org.apache.camel.support.service.ServiceSupport; + +/** Bounded recent history, lost when the context is destroyed. No persistent-storage guarantee. */ +public final class MemorySemanticAuditStore extends ServiceSupport implements SemanticAuditSink, SemanticAuditReader { + private final int capacity; + private final String epoch = UUID.randomUUID().toString(); + private final Deque<Entry> records = new ArrayDeque<>(); + private long sequence; + private long evicted; + + public MemorySemanticAuditStore(int capacity) { + if (capacity < 1 || capacity > 100000) { + throw new IllegalArgumentException("Invalid audit capacity"); + } + this.capacity = capacity; + } + + @Override + public synchronized void append(SemanticAuditRecord record) { + if (records.size() == capacity) { + records.removeLast(); + evicted++; + } + records.addFirst(new Entry(++sequence, record)); + } + + @Override + public synchronized Optional<SemanticAuditRecord> get(String eventId) { + return records.stream().map(Entry::record).filter(r -> r.getEventId().equals(eventId)).findFirst(); + } + + @Override + public synchronized Map<String, SemanticAuditRecord> getAll(List<String> eventIds) { + if (eventIds.size() > 100) { + throw new IllegalArgumentException("Too many audit evidence references"); + } + var selected = new HashSet<>(eventIds); + Map<String, SemanticAuditRecord> found = new LinkedHashMap<>(); + for (Entry entry : records) { + if (selected.remove(entry.record.getEventId())) { + found.put(entry.record.getEventId(), entry.record); + } + } + return Map.copyOf(found); + } + + @Override + public synchronized SemanticAuditPage query(SemanticAuditQuery query) { + long before = Long.MAX_VALUE; + if (query.getCursor() != null) { + String[] parts = query.getCursor().split(":", -1); + if (parts.length != 3 || !epoch.equals(parts[0]) + || !Integer.toString(query.getFilters().hashCode() + Objects.hashCode(query.getSince())).equals(parts[2])) { + throw new IllegalArgumentException("Audit cursor does not match this store or query"); + } + try { + before = Long.parseLong(parts[1]); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid audit cursor", e); + } + } + boolean expired = before != Long.MAX_VALUE && !records.isEmpty() && before <= records.getLast().sequence; + List<SemanticAuditRecord> page = new ArrayList<>(); + long last = 0; + boolean more = false; + for (Entry entry : records) { + if (entry.sequence < before && query.matches(entry.record)) { + if (page.size() == query.getLimit()) { + more = true; + break; + } + page.add(entry.record); + last = entry.sequence; + } + } + String next + = more ? epoch + ":" + last + ":" + (query.getFilters().hashCode() + Objects.hashCode(query.getSince())) : null; Review Comment: Fixed together with cursor validation in `fa99d7d1407991cf4c2e0bafbf9ce1e6f146a1aa`. Both paths call the same canonical-query fingerprint method. Pagination tests cover equivalent filter order, a changed timestamp boundary and a concrete pair of colliding Java map hashes. _AI-generated by OpenAI Codex on behalf of @luigidemasi via /oss-address-review._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAudit.java: ########## @@ -0,0 +1,515 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.semantic; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.builder.ThreadPoolBuilder; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Context-owned audit dispatcher and independent semantic observation hooks. */ +public final class SemanticAudit extends ServiceSupport { + public static final String REFERENCES = "CamelSemanticAuditReferences"; + public static final String REQUEST = "CamelSemanticAuditRequest"; + private static final Logger LOG = LoggerFactory.getLogger(SemanticAudit.class); + private static final Object CREATION_LOCK = new Object(); + private final CamelContext context; + private volatile SemanticAuditConfiguration configuration = SemanticAuditConfiguration.DISABLED; + private String configurationSource; + private final AtomicLong dropped = new AtomicLong(); + private final AtomicLong observerErrors = new AtomicLong(); + private final Map<String, AtomicLong> sinkErrors = new LinkedHashMap<>(); + private final Map<String, SemanticAuditSink> sinks = new LinkedHashMap<>(); + private volatile List<SemanticObserver> observers = List.of(); + private volatile ExecutorService executor; + private MemorySemanticAuditStore memory; + private volatile SemanticAuditReader reader; + private volatile boolean activated; + private boolean frozen; + private final Invocation noop = new Invocation(null, false); + + private SemanticAudit(CamelContext context) { + this.context = context; + } + + public static SemanticAudit get(CamelContext context) { + var extension = context.getCamelContextExtension(); + SemanticAudit audit = extension.getContextPlugin(SemanticAudit.class); + if (audit == null) { + synchronized (CREATION_LOCK) { + audit = extension.getContextPlugin(SemanticAudit.class); + if (audit == null) { + audit = new SemanticAudit(context); + extension.addContextPlugin(SemanticAudit.class, audit); + } + } + } + return audit; + } + + public synchronized void validateConfiguration(String source, SemanticAuditConfiguration candidate) { + if (configurationSource != null && !configurationSource.equals(source) && candidate != null) { + throw new IllegalArgumentException("Declare semantic audit configuration in only one resource"); + } + if (candidate == null && !source.equals(configurationSource)) { + return; + } + SemanticAuditConfiguration next = candidate == null ? SemanticAuditConfiguration.DISABLED : candidate; + if (frozen && !configuration.equals(next)) { + throw new IllegalArgumentException("Changing semantic audit configuration requires a context restart"); + } + } + + public synchronized void configure(String source, SemanticAuditConfiguration candidate) { + validateConfiguration(source, candidate); + if (candidate != null || source.equals(configurationSource)) { + configuration = candidate == null ? SemanticAuditConfiguration.DISABLED : candidate; + configurationSource = candidate == null ? null : source; + } + } + + synchronized void removeConfigurationSource(String source) { + if (source.equals(configurationSource)) { + configurationSource = null; + if (frozen) { + LOG.warn("Semantic audit resource {} was deleted; retaining the active configuration until context restart", + source); + } else { + configuration = SemanticAuditConfiguration.DISABLED; + } + } + } + + public void activate() { + if (!activated) { + synchronized (this) { + if (!activated) { + try { + context.addService(this, true, true); + activated = true; + } catch (Exception e) { + throw RuntimeCamelException.wrapRuntimeCamelException(e); + } + } + } + } + } + + @Override + protected void doStart() throws Exception { + synchronized (this) { + frozen = true; + } + observers = List.copyOf(context.getRegistry().findByType(SemanticObserver.class)); + if (memory == null) { + memory = new MemorySemanticAuditStore(configuration.getCapacity()); + } + for (String name : configuration.getSinks()) { + SemanticAuditSink sink = switch (name) { + case "memory" -> memory; + case "log" -> new LoggingSemanticAuditSink(); + default -> context.getRegistry().lookupByNameAndType(name, SemanticAuditSink.class); + }; + if (sink == null) { + throw new IllegalArgumentException("Unknown semantic audit sink: " + name); + } + sinks.put(name, sink); + sinkErrors.putIfAbsent(name, new AtomicLong()); + } + reader = "memory".equals(configuration.getReader()) + ? memory + : context.getRegistry().lookupByNameAndType(configuration.getReader(), SemanticAuditReader.class); + if (reader == null) { + throw new IllegalArgumentException("Unknown semantic audit reader: " + configuration.getReader()); + } + boolean capturing = configuration.isEnabled() || configuration.getExperts().containsValue(true); + if (!capturing) { + return; + } + try { + for (SemanticAuditSink sink : sinks.values()) { + ServiceHelper.startService(sink); + } + ServiceHelper.startService(reader); + executor = new ThreadPoolBuilder(context).poolSize(1).maxPoolSize(1) + .maxQueueSize(configuration.getQueueCapacity()).rejectedPolicy(ThreadPoolRejectedPolicy.Abort) + .build(this, "SemanticAudit"); + } catch (Exception e) { + ServiceHelper.stopService(sinks.values()); + ServiceHelper.stopService(reader); + throw e; + } + } + + @Override + protected void doStop() throws Exception { + ExecutorService current = executor; + executor = null; + if (current != null) { + current.shutdown(); + try { + if (!current.awaitTermination(5, TimeUnit.SECONDS)) { + dropped.addAndGet(current.shutdownNow().size()); + } + } catch (InterruptedException e) { + dropped.addAndGet(current.shutdownNow().size()); + Thread.currentThread().interrupt(); + } finally { + context.getExecutorServiceManager().shutdownNow(current); + } + } + ServiceHelper.stopService(sinks.values()); + ServiceHelper.stopService(reader); + } + + public synchronized SemanticAuditConfiguration getConfiguration() { + return configuration; + } + + public SemanticAuditReader getReader() { + activate(); + if (isStarted()) { + try { + ServiceHelper.startService(reader); + } catch (Exception failure) { + throw RuntimeCamelException.wrapRuntimeCamelException(failure); + } + } + return reader; + } + + public synchronized Map<String, Object> status() { + Map<String, Object> status = new LinkedHashMap<>(); + status.put("enabled", configuration.isEnabled()); + status.put("experts", configuration.getExperts()); + status.put("reader", configuration.getReader()); + status.put("sinks", configuration.getSinks()); + status.put("capacity", configuration.getCapacity()); + status.put("retained", memory == null ? 0 : memory.size()); + status.put("dropped", dropped.get()); + status.put("observerErrors", observerErrors.get()); + Map<String, Long> errors = new LinkedHashMap<>(); + sinkErrors.forEach((name, count) -> errors.put(name, count.get())); + status.put("sinkErrors", errors); + status.put("state", getStatus().name()); + status.put("observerCount", observers.size()); + status.put("openTelemetry", + ServiceHelper.isStarted(context.hasService( + service -> isOpenTelemetryTracer(service.getClass()) && ServiceHelper.isStarted(service))) + ? "active" : "inactive"); + return status; + } + + private static boolean isOpenTelemetryTracer(Class<?> type) { + for (Class<?> current = type; current != null; current = current.getSuperclass()) { + if (current.getName().equals("org.apache.camel.opentelemetry2.OpenTelemetryTracer") + || current.getName().equals("org.apache.camel.opentelemetry.OpenTelemetryTracer")) { + return true; + } + } + return false; + } + + /** Capture parent observation contexts on the caller thread before dispatching a console request. */ + public Request request(String origin) { + activate(); + Map<SemanticObserver, Object> parents = new IdentityHashMap<>(); + for (SemanticObserver observer : observers) { + try { + parents.put(observer, observer.captureContext()); + } catch (Exception | AssertionError | LinkageError failure) { + observerErrors.incrementAndGet(); + } + } + return new Request(origin, parents); + } + + /** A request identity is shared by cancellation observations and eventual provider completion. */ + public static final class Request { + private final String id = UUID.randomUUID().toString(); + private final String origin; + private final Map<SemanticObserver, Object> parents; + private final AtomicBoolean claimed = new AtomicBoolean(); + private volatile boolean audited; + + private Request(String origin, Map<SemanticObserver, Object> parents) { + this.origin = origin; + this.parents = parents; + } + } + + public Invocation begin( + Exchange exchange, String name, SemanticEvaluation evaluation, + SemanticCapabilities.Operation operation, String expert, String provider, String batchId, Request request) { + activate(); + if (request == null && exchange != null) { + request = exchange.getProperty(REQUEST, Request.class); + } + boolean first = request != null && request.claimed.compareAndSet(false, true); + boolean enabled = expert != null && configuration.isEnabled(expert); + if (request != null && enabled) { + request.audited = true; + } + if (!enabled && observers.isEmpty()) { + return noop; + } + Map<String, Object> fields = new LinkedHashMap<>(); + fields.put("schemaVersion", 1); + fields.put("eventId", UUID.randomUUID().toString()); + fields.put("invocationId", first ? request.id : UUID.randomUUID().toString()); + fields.put("timestamp", Instant.now().toString()); + fields.put("startedAt", fields.get("timestamp")); + fields.put("category", "evaluation"); + put(fields, "provider", provider); + fields.put("origin", request == null ? exchange == null ? "direct" : "route" : request.origin); + put(fields, "requestId", request == null ? null : request.id); + put(fields, "batchId", batchId); + put(fields, "definition", name); + put(fields, "target", name); + put(fields, "expert", expert); + put(fields, "operation", evaluation == null ? null : evaluation.getOperation()); + put(fields, "contextId", context.getName()); + if (exchange != null) { + put(fields, "exchangeId", exchange.getExchangeId()); + put(fields, "routeId", exchange.getFromRouteId()); + } + if (operation != null) { + Map<String, Object> semantics = new LinkedHashMap<>(); + put(semantics, "resultType", operation.getResultType().name()); + put(semantics, "meaning", operation.getResultMeaning()); + put(semantics, "probabilityMeaning", operation.getProbabilityMeaning()); + put(semantics, "confidenceMeaning", operation.getConfidenceMeaning()); + // Caller-authored score descriptions and arbitrary parameters are deliberately omitted. + fields.put("semantics", Map.copyOf(semantics)); + } + Invocation invocation = new Invocation(fields, enabled); + SemanticAuditRecord started = new SemanticAuditRecord(fields); + for (SemanticObserver observer : observers) { + try { + Object parent = request == null ? observer.captureContext() : request.parents.get(observer); + SemanticObserver.Observation handle = observer.started(started, parent); + if (handle != null) { + invocation.handles.add(handle); + } + } catch (Exception | AssertionError | LinkageError failure) { + observerErrors.incrementAndGet(); + } + } + return invocation; + } + + public final class Invocation { + private final Map<String, Object> fields; + private final boolean enabled; + private final long start = System.nanoTime(); + private final List<SemanticObserver.Observation> handles = new ArrayList<>(); + private boolean completed; Review Comment: I retained the plain boolean and documented the thread-confinement contract in `fa99d7d1407991cf4c2e0bafbf9ce1e6f146a1aa`. Start/completion run on the invocation thread; timeout/cancellation emits a separate request record and does not complete that invocation from another thread. Existing timeout/late-completion regressions exercise that separation. Volatile would not make the check-and-set atomic, so it would not be sufficient if concurrent completion were introduced. Invocation is now inside the internal runtime service rather than the supported route API. _AI-generated by OpenAI Codex on behalf of @luigidemasi via /oss-address-review._ ########## components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java: ########## @@ -686,21 +749,31 @@ private Object evaluate(Exchange exchange, boolean asPredicate) { invalid); } } + active.clear(); + failureReason = "provider_error"; if (batch) { Map<String, SemanticResult> results = new LinkedHashMap<>(); Map<String, Object> decisions = new LinkedHashMap<>(); for (Group group : current.groups) { if (Thread.currentThread().isInterrupted()) { throw new InterruptedException("Semantic batch evaluation interrupted"); } + active.clear(); + group.evaluations.keySet().forEach(name -> active.add(invocations.get(name))); + failureReason = "provider_error"; Map<String, SemanticResult> answers = group.expert.provider.evaluateBatch(group.evaluations, state); + failureReason = "invalid_result"; if (answers == null || !answers.keySet().equals(group.evaluations.keySet())) { throw new IllegalArgumentException("Semantic batch result names must match evaluation names"); } for (String name : group.evaluations.keySet()) { decisions.put(name, validateResult(name, group.expert, current.operations.get(name), answers.get(name))); } + for (String name : group.evaluations.keySet()) { + invocations.get(name).complete("success", "evaluation_completed", answers.get(name)); + } + active.clear(); Review Comment: Fixed in `fa99d7d1407991cf4c2e0bafbf9ce1e6f146a1aa`. After verifying the returned names, every answer is validated and audited independently; the first validation failure is still thrown, and no partial batch result or diagnostic map is published to the route. This also handles the reverse order: a valid answer after the invalid one is retained as success. The regression runs both refs:a,b and refs:b,a. Interleaving while immediately throwing would have left that second case misclassified. _AI-generated by OpenAI Codex on behalf of @luigidemasi via /oss-address-review._ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
