Skip to content

Commit 55761f9

Browse files
committed
agent: Add workers intakes and turn submission
Signed-off-by: AkaneTan <69606737+AkaneTan@users.noreply.github.com>
1 parent 392e0ca commit 55761f9

29 files changed

Lines changed: 2249 additions & 76 deletions

‎libraries/agent/build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,5 +8,6 @@ dependencies {
88
implementation(project(":libraries:internal"))
99

1010
testImplementation(project(":libraries:testkit"))
11+
testImplementation(libs.kotlinx.coroutines.test)
1112
testImplementation(libs.logback.classic)
1213
}

‎libraries/agent/src/main/kotlin/org/foedusprogramme/alexandrite/agent/AgentPlugin.kt‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,27 +2,45 @@ package org.foedusprogramme.alexandrite.agent
22

33
import kotlinx.coroutines.launch
44
import org.foedusprogramme.alexandrite.agent.config.AgentDirectory
5+
import org.foedusprogramme.alexandrite.agent.config.AgentSettings
56
import org.foedusprogramme.alexandrite.agent.model.Endpoints
7+
import org.foedusprogramme.alexandrite.agent.worker.ChatIntakes
8+
import org.foedusprogramme.alexandrite.agent.worker.TurnWorkers
69
import org.foedusprogramme.alexandrite.sdk.di.Lifecycle
710
import org.foedusprogramme.alexandrite.sdk.plugin.Plugin
811
import org.foedusprogramme.alexandrite.sdk.plugin.PluginScope
912
import org.slf4j.LoggerFactory
1013
import kotlin.coroutines.cancellation.CancellationException
14+
import kotlin.time.Duration.Companion.seconds
1115

1216
@Plugin(name = "Agent", description = "Turn pipeline, permissions and automation")
1317
public class AgentPlugin internal constructor(
18+
private val settings: AgentSettings,
1419
private val directory: AgentDirectory,
1520
private val endpoints: Endpoints,
21+
private val intakes: ChatIntakes,
22+
private val workers: TurnWorkers,
1623
private val scope: PluginScope,
1724
) : Lifecycle {
1825
override suspend fun onStart() {
1926
if (directory.agents.isEmpty()) logger.warn("No agent is configured at agent.agents: no chat is answered")
2027
}
2128

2229
override suspend fun onOpen() {
30+
workers.open()
2331
scope.launch { checkModels() }
2432
}
2533

34+
override suspend fun onClose() {
35+
workers.close()
36+
}
37+
38+
override suspend fun onDrain() {
39+
intakes.shutDown()
40+
val shutdown = settings.shutdown
41+
workers.drain(shutdown.turnGraceSeconds.seconds, shutdown.cancelJoinSeconds.seconds)
42+
}
43+
2644
/** Warns of each agent whose model its endpoint does not list. */
2745
private suspend fun checkModels() {
2846
for (agent in directory.agents) {

‎libraries/agent/src/main/kotlin/org/foedusprogramme/alexandrite/agent/config/AgentConfig.kt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,8 @@ internal class AgentConfig(
3535
val workspace: String? = null,
3636
/** The channel instances the agent serves, by `<type>:<name>`. */
3737
val channels: Map<String, AttachmentConfig> = emptyMap(),
38+
/** Groups of chat addresses that share one conversation, kept at the first chat of each. */
39+
val linkedChats: List<List<String>> = emptyList(),
3840
)
3941

4042
@Serializable

‎libraries/agent/src/main/kotlin/org/foedusprogramme/alexandrite/agent/config/AgentDirectory.kt‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@ internal class Agent(val id: AgentId, val config: AgentConfig, val attachments:
2020
val reasoning: ReasoningEffort? = config.reasoning?.let(ReasoningEffort::of)
2121

2222
val language: LanguageTag? = config.language?.let(LanguageTag::of)
23+
24+
val linkedChats: List<List<ChatAddress>> = config.linkedChats.map { group -> group.map(ChatAddress::parse) }
2325
}
2426

2527
/** An agent's attachment to a configured channel instance. */
@@ -35,19 +37,18 @@ internal class AgentDirectory(settings: AgentSettings, channels: ChannelDirector
3537
private val byId = agents.associateBy { it.id }
3638
private val attached = agents.flatMap { agent -> agent.attachments.map { it.instance to agent } }
3739
.groupBy({ it.first }, { it.second })
38-
private val homes = agents.flatMap { agent -> agent.attachments.mapNotNull { it.home?.to(agent) } }.toMap()
3940

4041
fun agent(id: AgentId): Agent? = byId[id]
4142

4243
/** The agents that serve [instance], in config order. */
4344
fun attached(instance: ChannelInstanceId): List<Agent> = attached[instance].orEmpty()
4445

4546
/** The agent that serves the chats of [instance] that choose no other, null when no agent serves it. */
46-
fun default(instance: ChannelInstanceId): Agent? =
47-
attached(instance).firstOrNull { agent -> agent.attachments.any { it.instance == instance && it.default } }
48-
49-
/** The agent whose home [chat] is, else the one whose home its parent is, null when it is no home chat. */
50-
fun home(chat: ChatAddress): Agent? = homes[chat] ?: homes[chat.parent]
47+
fun default(instance: ChannelInstanceId): Agent? {
48+
val agents = attached(instance)
49+
return agents.singleOrNull()
50+
?: agents.firstOrNull { agent -> agent.attachments.any { it.instance == instance && it.default } }
51+
}
5152
}
5253

5354
private fun resolved(id: AgentId, config: AgentConfig, channels: ChannelDirectory, endpoints: Endpoints): Agent {
@@ -85,6 +86,15 @@ private fun resolved(id: AgentId, config: AgentConfig, channels: ChannelDirector
8586
}
8687
}
8788
}
89+
val instances = attachments.map { it.instance }.toSet()
90+
for (chat in config.linkedChats.flatten().map(ChatAddress::parse).filter { it.instance !in instances }) {
91+
logger.warn(
92+
"Agent '{}' links chat {}, but no instance of channel type {} is configured: the chat gets no messages",
93+
id,
94+
chat,
95+
chat.instance.type,
96+
)
97+
}
8898
return Agent(id, config, attachments)
8999
}
90100

‎libraries/agent/src/main/kotlin/org/foedusprogramme/alexandrite/agent/config/ConfigChecks.kt‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,26 @@ private fun checkAgent(path: String, id: String, agent: AgentConfig) {
8080
require(agent.channels.isEmpty() || agent.model != null) {
8181
"$path.model is missing: an agent that serves channels needs a model"
8282
}
83+
checkLinks(path, agent)
84+
}
85+
86+
private fun checkLinks(path: String, agent: AgentConfig) {
87+
val attached = agent.channels.keys.mapNotNull(ChannelInstanceId::parseOrNull).toSet()
88+
val linked = mutableMapOf<ChatAddress, String>()
89+
for ((index, group) in agent.linkedChats.withIndex()) {
90+
require(group.size >= 2) { "$path.linkedChats[$index] holds fewer than two chats" }
91+
for ((position, text) in group.withIndex()) {
92+
val at = "linkedChats[$index][$position]"
93+
val chat = requireNotNull(ChatAddress.parseOrNull(text)) {
94+
"$path.$at is no chat address: write <type>:<name>:<chat>, such as \"telegram:work:123456\""
95+
}
96+
require(chat.instance in attached) {
97+
"$path.$at is a chat of ${chat.instance}, which the agent is not attached to: add it to $path.channels"
98+
}
99+
val first = linked.putIfAbsent(chat, at)
100+
require(first == null) { "$path.$at is the chat of $first again: a chat is in one group of an agent" }
101+
}
102+
}
83103
}
84104

85105
private fun checkGlobs(path: String, globs: List<String>) {
@@ -100,6 +120,7 @@ private fun checkDefaults(agents: Map<String, AgentConfig>) {
100120
agent.channels.map { (key, attachment) -> key to (id to attachment) }
101121
}
102122
for ((key, attachments) in attached.groupBy({ it.first }, { it.second })) {
123+
if (attachments.size == 1) continue
103124
val defaults = attachments.filter { it.second.default }.map { it.first }
104125
require(defaults.size == 1) {
105126
val agentsOf = attachments.joinToString { it.first }
Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package org.foedusprogramme.alexandrite.agent.control
2+
3+
import org.foedusprogramme.alexandrite.agent.worker.AgentTicket
4+
import org.foedusprogramme.alexandrite.sdk.chat.TurnId
5+
import org.foedusprogramme.alexandrite.sdk.di.Singleton
6+
import org.foedusprogramme.alexandrite.sdk.turn.Admission
7+
import org.foedusprogramme.alexandrite.sdk.turn.Submission
8+
import org.foedusprogramme.alexandrite.sdk.turn.TurnOutcome
9+
10+
/** Runs the command invocations that a handler claims, in a lane per chat. */
11+
@Singleton
12+
internal class CommandLanes {
13+
/** Placeholder until T2.5h builds the command lane. */
14+
fun submit(turn: TurnId, command: Submission.Command): Admission = ended(
15+
turn,
16+
TurnOutcome.Failed("Command '/${command.invocation.name}' was not run: commands run from T2.5h."),
17+
)
18+
19+
/** Placeholder until T2.5d answers a command that nobody claims and no message carried with a notice. */
20+
fun unknown(turn: TurnId): Admission = ended(turn, TurnOutcome.Completed(null))
21+
22+
private fun ended(turn: TurnId, outcome: TurnOutcome) = Admission.Accepted(AgentTicket.ended(turn, outcome))
23+
}

‎libraries/agent/src/main/kotlin/org/foedusprogramme/alexandrite/agent/control/SettingsStates.kt‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package org.foedusprogramme.alexandrite.agent.control
22

3-
import kotlinx.coroutines.runBlocking
43
import kotlinx.serialization.Serializable
54
import org.foedusprogramme.alexandrite.sdk.chat.AgentChatKey
65
import org.foedusprogramme.alexandrite.sdk.chat.AgentId
@@ -24,9 +23,15 @@ internal class SettingsStates(states: ChatStates) {
2423
private val language: ChatState<AgentChatKey, Choice<LanguageTag>> = states.agentState(LANGUAGE, Choice())
2524
private val selections = ConcurrentHashMap<ChatAddress, Choice<AgentId>>()
2625

27-
/** The agent that [chat] or its parent switched to, read with a blocking call the first time. */
28-
fun selectedAgent(chat: ChatAddress): AgentId? =
29-
(selections[chat] ?: runBlocking { agent.get(chat) }.also { selections[chat] = it }).value
26+
/** The agent that [chat] or its parent switched to, null until [loadAgent] has read it. */
27+
fun cachedAgent(chat: ChatAddress): Choice<AgentId>? = selections[chat]
28+
29+
/** Reads the agent that [chat] or its parent switched to, once per chat. */
30+
suspend fun loadAgent(chat: ChatAddress): Choice<AgentId> {
31+
selections[chat]?.let { return it }
32+
val read = agent.get(chat)
33+
return selections.putIfAbsent(chat, read) ?: read
34+
}
3035

3136
suspend fun model(key: AgentChatKey): ModelRef? = model.get(key).value
3237

‎libraries/agent/src/main/kotlin/org/foedusprogramme/alexandrite/agent/routing/ChatRouter.kt‎

Lines changed: 33 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package org.foedusprogramme.alexandrite.agent.routing
22

33
import org.foedusprogramme.alexandrite.agent.config.Agent
44
import org.foedusprogramme.alexandrite.agent.config.AgentDirectory
5+
import org.foedusprogramme.alexandrite.agent.control.Choice
56
import org.foedusprogramme.alexandrite.agent.control.SettingsStates
67
import org.foedusprogramme.alexandrite.sdk.channel.ChannelDirectory
78
import org.foedusprogramme.alexandrite.sdk.chat.AgentChatKey
@@ -13,7 +14,8 @@ import java.util.concurrent.ConcurrentHashMap
1314

1415
/** Which agent serves a chat. */
1516
internal sealed interface Resolution {
16-
data class Served(val key: AgentChatKey) : Resolution
17+
/** Served at [key] for the chat [origin], which the turn came from. */
18+
data class Served(val key: AgentChatKey, val origin: ChatAddress) : Resolution
1719

1820
/** The chat's channel instance is not configured. */
1921
data object Unknown : Resolution
@@ -22,33 +24,51 @@ internal sealed interface Resolution {
2224
data object NoAgent : Resolution
2325
}
2426

25-
/** Finds the agent of a chat: its home agent, else the agent the chat switched to, else its instance's default. */
27+
/** Finds the agent of a chat: the one it switched to, else its home agent, else its instance's default. */
2628
@Singleton
2729
internal class ChatRouter(
2830
private val channels: ChannelDirectory,
2931
private val directory: AgentDirectory,
32+
private val topology: ChatTopology,
3033
private val settings: SettingsStates,
3134
) {
3235
private val warned = ConcurrentHashMap.newKeySet<ChatAddress>()
3336

34-
fun resolve(chat: ChatAddress): Resolution {
35-
if (chat.instance !in channels.instances) return Resolution.Unknown
37+
/** How [chat] resolves from memory, null while the agent it switched to is not read yet. */
38+
fun cached(chat: ChatAddress): Resolution? {
39+
unserved(chat)?.let { return it }
3640
val attached = directory.attached(chat.instance)
37-
if (attached.isEmpty()) return Resolution.NoAgent
38-
val agent = directory.home(chat)?.id
39-
?: selected(chat, attached)
41+
val choice = if (attached.size < 2) Choice() else settings.cachedAgent(chat) ?: return null
42+
return served(chat, attached, choice.value)
43+
}
44+
45+
/** How [chat] resolves, reading the agent it switched to the first time. */
46+
suspend fun resolve(chat: ChatAddress): Resolution {
47+
unserved(chat)?.let { return it }
48+
val attached = directory.attached(chat.instance)
49+
val choice = if (attached.size < 2) Choice() else settings.loadAgent(chat)
50+
return served(chat, attached, choice.value)
51+
}
52+
53+
private fun unserved(chat: ChatAddress): Resolution? = when {
54+
chat.instance !in channels.instances -> Resolution.Unknown
55+
directory.attached(chat.instance).isEmpty() -> Resolution.NoAgent
56+
else -> null
57+
}
58+
59+
private fun served(chat: ChatAddress, attached: List<Agent>, selected: AgentId?): Resolution.Served {
60+
val agent = selected?.let { attachedOrWarn(chat, attached, it) }
61+
?: topology.homeAgent(chat)
4062
?: checkNotNull(directory.default(chat.instance)).id
41-
return Resolution.Served(AgentChatKey(agent, chat))
63+
return Resolution.Served(topology.key(agent, chat), chat)
4264
}
4365

44-
/** The agent [chat] switched to, null when it is none of [attached] or only one agent is attached. */
45-
private fun selected(chat: ChatAddress, attached: List<Agent>): AgentId? {
46-
if (attached.size < 2) return null
47-
val selected = settings.selectedAgent(chat) ?: return null
66+
/** [selected], null when it no longer serves [chat]'s instance. */
67+
private fun attachedOrWarn(chat: ChatAddress, attached: List<Agent>, selected: AgentId): AgentId? {
4868
if (attached.any { it.id == selected }) return selected
4969
if (warned.add(chat)) {
5070
logger.warn(
51-
"Chat {} switched to agent '{}', which no longer serves {}: the chat goes to its default agent",
71+
"Chat {} switched to agent '{}', which no longer serves {}: the chat goes to its home or default agent",
5272
chat,
5373
selected,
5474
chat.instance,
Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
package org.foedusprogramme.alexandrite.agent.routing
2+
3+
import org.foedusprogramme.alexandrite.agent.config.AgentDirectory
4+
import org.foedusprogramme.alexandrite.sdk.chat.AgentChatKey
5+
import org.foedusprogramme.alexandrite.sdk.chat.AgentId
6+
import org.foedusprogramme.alexandrite.sdk.chat.ChannelInstanceId
7+
import org.foedusprogramme.alexandrite.sdk.chat.ChatAddress
8+
import org.foedusprogramme.alexandrite.sdk.di.Singleton
9+
10+
/** Chats of one agent that share the conversation kept at [anchor]. */
11+
internal data class LinkGroup(val anchor: ChatAddress, val members: List<ChatAddress>) {
12+
val chats: List<ChatAddress> get() = listOf(anchor) + members
13+
}
14+
15+
/**
16+
* The home chats and linked groups of the agents, read without suspending.
17+
*
18+
* A change that breaks a rule throws [IllegalArgumentException] with a message for the user and changes nothing.
19+
*/
20+
@Singleton
21+
internal class ChatTopology(private val directory: AgentDirectory) {
22+
private val lock = Any()
23+
24+
@Volatile
25+
private var current = Topology(
26+
directory.agents.associate { agent ->
27+
agent.id to
28+
agent.attachments.mapNotNull { attachment -> attachment.home?.let { attachment.instance to it } }
29+
.toMap()
30+
},
31+
directory.agents.associate { agent ->
32+
agent.id to agent.linkedChats.map { LinkGroup(it.first(), it.drop(1)) }
33+
},
34+
)
35+
36+
/** The agent whose home [chat] is, else the one whose home its parent is, null when it is no home chat. */
37+
fun homeAgent(chat: ChatAddress): AgentId? = current.let { it.homeAgents[chat] ?: it.homeAgents[chat.parent] }
38+
39+
fun home(agent: AgentId, instance: ChannelInstanceId): ChatAddress? = current.homes[agent]?.get(instance)
40+
41+
fun groups(agent: AgentId): List<LinkGroup> = current.groups[agent].orEmpty()
42+
43+
/** The group of [agent] that holds [chat], null when none does. */
44+
fun group(agent: AgentId, chat: ChatAddress): LinkGroup? = current.groupOf[agent]?.get(chat)
45+
46+
/** Where [agent] keeps its conversation with [chat]. */
47+
fun key(agent: AgentId, chat: ChatAddress): AgentChatKey = AgentChatKey(agent, group(agent, chat)?.anchor ?: chat)
48+
49+
/** Makes [chat] the home of [agent] on [instance], or leaves the instance without one when [chat] is null. */
50+
fun setHome(agent: AgentId, instance: ChannelInstanceId, chat: ChatAddress?) {
51+
synchronized(lock) {
52+
val topology = current
53+
requireAttached(agent, instance)
54+
if (chat != null) {
55+
require(chat.instance == instance) { "$chat is no chat of $instance." }
56+
val other = topology.homeAgents[chat]
57+
require(other == null || other == agent) { "$chat is the home chat of agent '$other'." }
58+
}
59+
val homes = topology.homes[agent].orEmpty().toMutableMap()
60+
if (chat == null) homes.remove(instance) else homes[instance] = chat
61+
current = Topology(topology.homes + (agent to homes), topology.groups)
62+
}
63+
}
64+
65+
/** Links [member] into the group of [chat], which [chat] anchors when it is in none, and returns that group. */
66+
fun link(agent: AgentId, chat: ChatAddress, member: ChatAddress): LinkGroup = synchronized(lock) {
67+
val topology = current
68+
requireAttached(agent, chat.instance)
69+
requireAttached(agent, member.instance)
70+
require(member != chat) { "A chat cannot be linked to itself." }
71+
val linked = topology.groupOf[agent]?.get(member)
72+
require(linked == null) { "$member is linked to ${linked?.anchor} already: unlink it first." }
73+
val group = topology.groupOf[agent]?.get(chat)
74+
val grown = group?.copy(members = group.members + member) ?: LinkGroup(chat, listOf(member))
75+
val groups = topology.groups[agent].orEmpty()
76+
val next = if (group == null) groups + grown else groups.map { if (it == group) grown else it }
77+
current = Topology(topology.homes, topology.groups + (agent to next))
78+
grown
79+
}
80+
81+
/** Takes [member] out of its group of [agent], and returns whether it was in one. */
82+
fun unlink(agent: AgentId, member: ChatAddress): Boolean = synchronized(lock) {
83+
val topology = current
84+
val group = topology.groupOf[agent]?.get(member) ?: return false
85+
require(group.anchor != member) { "$member anchors its group: unlink the other chats first." }
86+
val shrunk = group.copy(members = group.members - member)
87+
val next = topology.groups.getValue(agent).mapNotNull {
88+
if (it != group) it else shrunk.takeIf { shrunk.members.isNotEmpty() }
89+
}
90+
current = Topology(topology.homes, topology.groups + (agent to next))
91+
true
92+
}
93+
94+
private fun requireAttached(agent: AgentId, instance: ChannelInstanceId) {
95+
require(directory.agent(agent)?.attachments.orEmpty().any { it.instance == instance }) {
96+
"Agent '$agent' does not serve $instance."
97+
}
98+
}
99+
}
100+
101+
private class Topology(
102+
val homes: Map<AgentId, Map<ChannelInstanceId, ChatAddress>>,
103+
val groups: Map<AgentId, List<LinkGroup>>,
104+
) {
105+
val homeAgents: Map<ChatAddress, AgentId> =
106+
homes.flatMap { (agent, byInstance) -> byInstance.values.map { it to agent } }.toMap()
107+
108+
val groupOf: Map<AgentId, Map<ChatAddress, LinkGroup>> =
109+
groups.mapValues { (_, groups) -> groups.flatMap { group -> group.chats.map { it to group } }.toMap() }
110+
}

0 commit comments

Comments
 (0)