File Extension/Tasks/Server/Store/InMemoryTaskStore.php

1:	<?php
2:
3:	declare(strict_types=1);
4:
5:	/**
6:	* This file is part of the Nexus MCP SDK package.
7:	*
8:	* (c) 2026 John Paul E. Balandan, CPA <paulbalandan@gmail.com>
9:	*
10:	* For the full copyright and license information, please view
11:	* the LICENSE file that was distributed with this source code.
12:	*/
13:
14:	namespace Nexus\Mcp\Extension\Tasks\Server\Store;
15:
16:	use Nexus\Assert\Assert;
17:	use Nexus\Clock\Clock;
18:	use Nexus\Clock\SystemClock;
19:	use Nexus\Mcp\Core\Exception\RuntimeException;
20:	use Nexus\Mcp\Core\JsonRpc\ErrorFactory;
21:	use Nexus\Mcp\Core\Schema\Enum\ProtocolErrorCode;
22:	use Nexus\Mcp\Core\Schema\Request\InputRequest;
23:	use Nexus\Mcp\Core\Schema\Result\InputResponse;
24:	use Nexus\Mcp\Extension\Tasks\Schema\Enum\TaskStatus;
25:	use Nexus\Mcp\Extension\Tasks\Server\Exception\InputRequestKeyReusedException;
26:
27:	/**
28:	* In-memory implementation of `TaskStoreInterface`.
29:	*/
30:	final class InMemoryTaskStore implements TaskStoreInterface
31:	{
32:	public const int DEFAULT_MAX_RECORDS = 10_000;
33:
34:	/**
35:	* @var array<non-empty-string, TaskRecord>
36:	*/
37:	private array $records = [];
38:
39:	/**
40:	* Settle instants in settle order, so the first entry is the oldest settled record.
41:	*
42:	* @var array<non-empty-string, \DateTimeImmutable>
43:	*/
44:	private array $terminalAt = [];
45:
46:	/**
47:	* @param int<1, max> $maxRecords Records held at once, settled ones included
48:	*/
49:	public function __construct(
50:	private readonly Clock $clock = new SystemClock(),
51:	private readonly int $maxRecords = self::DEFAULT_MAX_RECORDS,
52:	) {
53:	Assert::that($maxRecords)->isPositiveInt('maxRecords must be a positive integer, {value} given.');
54:	}
55:
56:	#[\Override]
57:	public function createTask(string $toolName, ?array $arguments, ?int $ttlMs, int $pollIntervalMs): TaskRecord
58:	{
59:	$instant = $this->clock->now();
60:	$this->reclaim($this->convertToMilliseconds($instant));
61:
62:	$now = $instant->format(\DateTimeInterface::ATOM);
63:	$taskId = bin2hex(random_bytes(16));
64:
65:	$record = new TaskRecord(
66:	taskId: $taskId,
67:	toolName: $toolName,
68:	status: TaskStatus::Working,
69:	createdAt: $now,
70:	lastUpdatedAt: $now,
71:	ttlMs: $ttlMs,
72:	pollIntervalMs: $pollIntervalMs,
73:	arguments: $arguments,
74:	);
75:
76:	$this->records[$taskId] = $record;
77:
78:	return $record;
79:	}
80:
81:	#[\Override]
82:	public function findTask(string $taskId): ?TaskRecord
83:	{
84:	return $this->resolveTask($taskId, null);
85:	}
86:
87:	#[\Override]
88:	public function trySetWorking(string $taskId): bool
89:	{
90:	$record = $this->findLive($taskId);
91:
92:	if (null === $record) {
93:	return false;
94:	}
95:
96:	$this->replaceRecord($record, TaskStatus::Working, ['pendingInputRequests' => []]);
97:
98:	return true;
99:	}
100:
101:	#[\Override]
102:	public function trySetCompleted(string $taskId, array $result): bool
103:	{
104:	$record = $this->findLive($taskId);
105:
106:	if (null === $record) {
107:	return false;
108:	}
109:
110:	$this->replaceRecord($record, TaskStatus::Completed, [
111:	'result' => $result,
112:	'pendingInputRequests' => [],
113:	'inputResponses' => [],
114:	'requestState' => null,
115:	]);
116:
117:	return true;
118:	}
119:
120:	#[\Override]
121:	public function trySetFailed(string $taskId, array $error, ?string $statusMessage = null): bool
122:	{
123:	$record = $this->findLive($taskId);
124:
125:	if (null === $record) {
126:	return false;
127:	}
128:
129:	$this->replaceRecord($record, TaskStatus::Failed, [
130:	'error' => $error,
131:	'statusMessage' => $statusMessage,
132:	'pendingInputRequests' => [],
133:	'inputResponses' => [],
134:	'requestState' => null,
135:	]);
136:
137:	return true;
138:	}
139:
140:	#[\Override]
141:	public function trySetCancelled(string $taskId): bool
142:	{
143:	$record = $this->findLive($taskId);
144:
145:	if (null === $record) {
146:	return false;
147:	}
148:
149:	$this->replaceRecord($record, TaskStatus::Cancelled, [
150:	'pendingInputRequests' => [],
151:	'inputResponses' => [],
152:	'requestState' => null,
153:	]);
154:
155:	return true;
156:	}
157:
158:	#[\Override]
159:	public function trySetInputRequired(string $taskId, array $inputRequests, ?string $requestState): bool
160:	{
161:	$record = $this->findLive($taskId);
162:
163:	if (null === $record) {
164:	return false;
165:	}
166:
167:	foreach (array_keys($inputRequests) as $key) {
168:	if (isset($record->issuedInputKeys[$key])) {
169:	throw new InputRequestKeyReusedException($taskId, $key);
170:	}
171:	}
172:
173:	$this->replaceRecord($record, TaskStatus::InputRequired, [
174:	'pendingInputRequests' => $inputRequests,
175:	'requestState' => $requestState,
176:	'issuedInputKeys' => $record->issuedInputKeys + array_fill_keys(array_keys($inputRequests), true),
177:	]);
178:
179:	return true;
180:	}
181:
182:	#[\Override]
183:	public function resolveInputRequests(string $taskId, array $inputResponses): ?TaskRecord
184:	{
185:	$record = $this->findTask($taskId);
186:
187:	if (null === $record) {
188:	return null;
189:	}
190:
191:	$pending = $record->pendingInputRequests;
192:	$accepted = $record->inputResponses;
193:	$changed = false;
194:
195:	foreach ($inputResponses as $key => $response) {
196:	if (! \array_key_exists($key, $pending)) {
197:	continue;
198:	}
199:
200:	unset($pending[$key]);
201:	$accepted[$key] = $response;
202:	$changed = true;
203:	}
204:
205:	if (! $changed) {
206:	return $record;
207:	}
208:
209:	return $this->replaceRecord($record, $record->status, [
210:	'pendingInputRequests' => $pending,
211:	'inputResponses' => $accepted,
212:	]);
213:	}
214:
215:	/**
216:	* Frees room for one more record: drops the settled records that have expired in settle order, and at the
217:	* ceiling resolves every record once and then evicts the oldest settled one.
218:	*
219:	* @throws RuntimeException
220:	*/
221:	private function reclaim(int $nowMs): void
222:	{
223:	foreach (array_keys($this->terminalAt) as $taskId) {
224:	$record = $this->records[$taskId] ?? null;
225:	\assert($record instanceof TaskRecord);
226:
227:	if (! $this->hasExpired($taskId, $record, $nowMs)) {
228:	break;
229:	}
230:
231:	unset($this->records[$taskId], $this->terminalAt[$taskId]);
232:	}
233:
234:	if ($this->maxRecords > \count($this->records)) {
235:	return;
236:	}
237:
238:	foreach (array_keys($this->records) as $taskId) {
239:	$this->resolveTask($taskId, $nowMs);
240:	}
241:
242:	if ($this->maxRecords > \count($this->records)) {
243:	return;
244:	}
245:
246:	$oldest = array_key_first($this->terminalAt);
247:
248:	if (null === $oldest) {
249:	throw new RuntimeException(\sprintf('The task store holds its maximum of %d records and none of them has settled.', $this->maxRecords));
250:	}
251:
252:	unset($this->records[$oldest], $this->terminalAt[$oldest]);
253:	}
254:
255:	/**
256:	* `findTask()` with the current time in epoch milliseconds precomputed, or `null` to read the clock on demand.
257:	*
258:	* @param non-empty-string $taskId
259:	*/
260:	private function resolveTask(string $taskId, ?int $nowMs): ?TaskRecord
261:	{
262:	$record = $this->records[$taskId] ?? null;
263:
264:	if (null === $record) {
265:	return null;
266:	}
267:
268:	if ($this->hasExpired($taskId, $record, $nowMs)) {
269:	unset($this->records[$taskId], $this->terminalAt[$taskId]);
270:
271:	return null;
272:	}
273:
274:	if ($this->hasOverstayed($record, $nowMs)) {
275:	return $this->replaceRecord($record, TaskStatus::Failed, [
276:	'error' => ErrorFactory::create(ProtocolErrorCode::InternalError, 'The task did not settle within its ttl.')->toArray(),
277:	'pendingInputRequests' => [],
278:	'inputResponses' => [],
279:	'requestState' => null,
280:	]);
281:	}
282:
283:	return $record;
284:	}
285:
286:	/**
287:	* @param array{
288:	* result?: null\|array<string, mixed>,
289:	* error?: null\|array<string, mixed>,
290:	* pendingInputRequests?: array<int\|non-empty-string, InputRequest>,
291:	* inputResponses?: array<int\|non-empty-string, InputResponse>,
292:	* requestState?: null\|string,
293:	* issuedInputKeys?: array<array-key, true>,
294:	* statusMessage?: null\|non-empty-string,
295:	* } $changes
296:	*/
297:	private function replaceRecord(TaskRecord $record, TaskStatus $status, array $changes): TaskRecord
298:	{
299:	$instant = $this->clock->now();
300:
301:	$updated = new TaskRecord(
302:	taskId: $record->taskId,
303:	toolName: $record->toolName,
304:	status: $status,
305:	createdAt: $record->createdAt,
306:	lastUpdatedAt: $instant->format(\DateTimeInterface::ATOM),
307:	ttlMs: $record->ttlMs,
308:	pollIntervalMs: $record->pollIntervalMs,
309:	arguments: $record->arguments,
310:	result: \array_key_exists('result', $changes) ? $changes['result'] : $record->result,
311:	error: \array_key_exists('error', $changes) ? $changes['error'] : $record->error,
312:	pendingInputRequests: $changes['pendingInputRequests'] ?? $record->pendingInputRequests,
313:	inputResponses: $changes['inputResponses'] ?? $record->inputResponses,
314:	requestState: \array_key_exists('requestState', $changes) ? $changes['requestState'] : $record->requestState,
315:	issuedInputKeys: $changes['issuedInputKeys'] ?? $record->issuedInputKeys,
316:	statusMessage: \array_key_exists('statusMessage', $changes) ? $changes['statusMessage'] : $record->statusMessage,
317:	);
318:
319:	$this->records[$record->taskId] = $updated;
320:
321:	if ($this->isTerminal($status)) {
322:	$this->terminalAt[$record->taskId] = $instant;
323:	}
324:
325:	return $updated;
326:	}
327:
328:	/**
329:	* The record for `$taskId` when it exists, is unexpired, and is not terminal.
330:	*
331:	* @param non-empty-string $taskId
332:	*/
333:	private function findLive(string $taskId): ?TaskRecord
334:	{
335:	$record = $this->findTask($taskId);
336:
337:	if (null === $record) {
338:	return null;
339:	}
340:
341:	return $this->isTerminal($record->status) ? null : $record;
342:	}
343:
344:	/**
345:	* @param non-empty-string $taskId
346:	*/
347:	private function hasExpired(string $taskId, TaskRecord $record, ?int $nowMs): bool
348:	{
349:	if (null === $record->ttlMs) {
350:	return false;
351:	}
352:
353:	$terminalAt = $this->terminalAt[$taskId] ?? null;
354:
355:	if (null === $terminalAt) {
356:	return false;
357:	}
358:
359:	$nowMs ??= $this->convertToMilliseconds($this->clock->now());
360:
361:	return $record->ttlMs <= $nowMs - $this->convertToMilliseconds($terminalAt);
362:	}
363:
364:	/**
365:	* True when a non-terminal record has outlived `createdAt + ttlMs`, which SEP-2663 allows failing.
366:	*/
367:	private function hasOverstayed(TaskRecord $record, ?int $nowMs): bool
368:	{
369:	if (null === $record->ttlMs \|\| $this->isTerminal($record->status)) {
370:	return false;
371:	}
372:
373:	$nowMs ??= $this->convertToMilliseconds($this->clock->now());
374:
375:	return $record->ttlMs <= $nowMs - $this->convertToMilliseconds(new \DateTimeImmutable($record->createdAt));
376:	}
377:
378:	private function isTerminal(TaskStatus $status): bool
379:	{
380:	return match ($status) {
381:	TaskStatus::Completed, TaskStatus::Cancelled, TaskStatus::Failed => true,
382:	TaskStatus::Working, TaskStatus::InputRequired => false,
383:	};
384:	}
385:
386:	private function convertToMilliseconds(\DateTimeImmutable $instant): int
387:	{
388:	return $instant->getTimestamp() * 1_000 + intdiv((int) $instant->format('u'), 1_000);
389:	}
390:	}
391:

Namespaces

Classes

Interfaces