mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-17 12:54:40 +08:00
feat(agent,trigger): wire agent_lifecycle as a real event source
This commit is contained in:
parent
dae74cf26c
commit
231724d6a9
@ -3,12 +3,15 @@ package vip.mate.agent;
|
|||||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.context.ApplicationEventPublisher;
|
||||||
import org.springframework.context.event.EventListener;
|
import org.springframework.context.event.EventListener;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
import org.springframework.util.StringUtils;
|
import org.springframework.util.StringUtils;
|
||||||
import reactor.core.publisher.Flux;
|
import reactor.core.publisher.Flux;
|
||||||
import vip.mate.agent.context.ChatOrigin;
|
import vip.mate.agent.context.ChatOrigin;
|
||||||
import vip.mate.agent.context.ChatOriginHolder;
|
import vip.mate.agent.context.ChatOriginHolder;
|
||||||
|
import vip.mate.agent.event.AgentLifecycleEvent;
|
||||||
import vip.mate.agent.model.AgentEntity;
|
import vip.mate.agent.model.AgentEntity;
|
||||||
import vip.mate.agent.repository.AgentMapper;
|
import vip.mate.agent.repository.AgentMapper;
|
||||||
import vip.mate.exception.MateClawException;
|
import vip.mate.exception.MateClawException;
|
||||||
@ -43,6 +46,11 @@ public class AgentService {
|
|||||||
private final MemoryLifecycleMediator lifecycleMediator;
|
private final MemoryLifecycleMediator lifecycleMediator;
|
||||||
private final MemoryProperties memoryProperties;
|
private final MemoryProperties memoryProperties;
|
||||||
|
|
||||||
|
/** Field-injected publisher for agent_lifecycle trigger events; the
|
||||||
|
* trigger module's bridge listens and forwards into ingest. */
|
||||||
|
@Autowired(required = false)
|
||||||
|
private ApplicationEventPublisher events;
|
||||||
|
|
||||||
/** 运行时 Agent 实例缓存(agentId -> BaseAgent) */
|
/** 运行时 Agent 实例缓存(agentId -> BaseAgent) */
|
||||||
private final Map<Long, BaseAgent> agentInstances = new ConcurrentHashMap<>();
|
private final Map<Long, BaseAgent> agentInstances = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
@ -76,18 +84,51 @@ public class AgentService {
|
|||||||
agent.setAgentType("react");
|
agent.setAgentType("react");
|
||||||
}
|
}
|
||||||
agentMapper.insert(agent);
|
agentMapper.insert(agent);
|
||||||
|
publishLifecycle(agent, "spawned");
|
||||||
return agent;
|
return agent;
|
||||||
}
|
}
|
||||||
|
|
||||||
public AgentEntity updateAgent(AgentEntity agent) {
|
public AgentEntity updateAgent(AgentEntity agent) {
|
||||||
|
// Detect enabled-flag flip so the lifecycle event reflects the
|
||||||
|
// intent rather than every metadata edit. Reading the prior row
|
||||||
|
// is cheap and gives us a clean diff source.
|
||||||
|
AgentEntity prior = agentMapper.selectById(agent.getId());
|
||||||
agentMapper.updateById(agent);
|
agentMapper.updateById(agent);
|
||||||
agentInstances.remove(agent.getId());
|
agentInstances.remove(agent.getId());
|
||||||
|
if (prior != null && prior.getEnabled() != null
|
||||||
|
&& !prior.getEnabled().equals(agent.getEnabled())) {
|
||||||
|
publishLifecycle(agent,
|
||||||
|
Boolean.TRUE.equals(agent.getEnabled()) ? "enabled" : "disabled");
|
||||||
|
}
|
||||||
return agent;
|
return agent;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void deleteAgent(Long id) {
|
public void deleteAgent(Long id) {
|
||||||
|
AgentEntity prior = agentMapper.selectById(id);
|
||||||
agentMapper.deleteById(id);
|
agentMapper.deleteById(id);
|
||||||
agentInstances.remove(id);
|
agentInstances.remove(id);
|
||||||
|
if (prior != null) publishLifecycle(prior, "terminated");
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Best-effort publish of an {@link AgentLifecycleEvent}. A publish
|
||||||
|
* failure must never roll back the agent CRUD that just succeeded —
|
||||||
|
* the agent_lifecycle trigger surface is observability, not the
|
||||||
|
* canonical record.
|
||||||
|
*/
|
||||||
|
private void publishLifecycle(AgentEntity agent, String phase) {
|
||||||
|
if (events == null || agent == null) return;
|
||||||
|
try {
|
||||||
|
events.publishEvent(new AgentLifecycleEvent(
|
||||||
|
agent.getWorkspaceId() == null ? 0L : agent.getWorkspaceId(),
|
||||||
|
agent.getId() == null ? 0L : agent.getId(),
|
||||||
|
agent.getName(),
|
||||||
|
phase,
|
||||||
|
System.currentTimeMillis()));
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("[AgentService] lifecycle publish failed for agent {} ({}): {}",
|
||||||
|
agent.getId(), phase, e.getMessage());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@ -0,0 +1,25 @@
|
|||||||
|
package vip.mate.agent.event;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Spring application event fired when an agent's lifecycle state changes.
|
||||||
|
* The trigger module subscribes via {@code @EventListener} and forwards
|
||||||
|
* the payload through {@code TriggerEventIngestService} so triggers of
|
||||||
|
* pattern type {@code agent_lifecycle} can fan out to workflows.
|
||||||
|
*
|
||||||
|
* <p>{@code phase} matches the matcher's vocabulary: {@code spawned} for
|
||||||
|
* a fresh create, {@code enabled} / {@code disabled} for a flag flip,
|
||||||
|
* {@code terminated} for a delete. {@code crashed} is reserved for v1
|
||||||
|
* once the agent runtime grows a structured error hook.
|
||||||
|
*
|
||||||
|
* <p>The dedup key downstream is {@code phase + ":" + agentId + ":" +
|
||||||
|
* timestamp}; that's stable across retries of the same operation but
|
||||||
|
* lets the same agent flip enabled/disabled repeatedly without the
|
||||||
|
* trigger pipeline collapsing the events.
|
||||||
|
*/
|
||||||
|
public record AgentLifecycleEvent(
|
||||||
|
long workspaceId,
|
||||||
|
long agentId,
|
||||||
|
String agentName,
|
||||||
|
String phase,
|
||||||
|
long timestamp
|
||||||
|
) {}
|
||||||
@ -0,0 +1,55 @@
|
|||||||
|
package vip.mate.trigger.dispatch;
|
||||||
|
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.context.event.EventListener;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
import vip.mate.agent.event.AgentLifecycleEvent;
|
||||||
|
import vip.mate.trigger.ingest.TriggerEventEnvelope;
|
||||||
|
import vip.mate.trigger.ingest.TriggerEventIngestService;
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Forwards {@link AgentLifecycleEvent} into the trigger pipeline as
|
||||||
|
* {@code agent_lifecycle} envelopes. Lives in the trigger module so the
|
||||||
|
* agent runtime stays free of trigger / ingest dependencies, matching
|
||||||
|
* the workflow_completion + channel_message bridge pattern.
|
||||||
|
*
|
||||||
|
* <p>The dedup key composes phase + agentId + timestamp so the same
|
||||||
|
* agent flipping enabled / disabled repeatedly stays observable, but
|
||||||
|
* an at-least-once retry of the same exact lifecycle event collapses.
|
||||||
|
* Failures inside ingest are logged and swallowed.
|
||||||
|
*/
|
||||||
|
@Slf4j
|
||||||
|
@Component
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
public class AgentLifecycleEventBridge {
|
||||||
|
|
||||||
|
private final TriggerEventIngestService ingestService;
|
||||||
|
|
||||||
|
@EventListener
|
||||||
|
public void onLifecycle(AgentLifecycleEvent event) {
|
||||||
|
if (event == null) return;
|
||||||
|
try {
|
||||||
|
Map<String, Object> data = new HashMap<>();
|
||||||
|
// The matcher reads `agentId` and `phase` out of the envelope
|
||||||
|
// data; the field names mirror the matcher's vocabulary so
|
||||||
|
// pattern_json can narrow precisely.
|
||||||
|
data.put("agentId", event.agentId());
|
||||||
|
if (event.agentName() != null) data.put("agentName", event.agentName());
|
||||||
|
data.put("phase", event.phase());
|
||||||
|
data.put("timestamp", event.timestamp());
|
||||||
|
ingestService.ingest(new TriggerEventEnvelope(
|
||||||
|
event.workspaceId(),
|
||||||
|
"agent_lifecycle",
|
||||||
|
event.phase() + ":" + event.agentId() + ":" + event.timestamp(),
|
||||||
|
"system",
|
||||||
|
data));
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("[AgentLifecycleBridge] forwarding agent {} phase={} failed: {}",
|
||||||
|
event.agentId(), event.phase(), e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue
Block a user